OpenLineage マッピング

Data Lineage API は、リネージ収集のオープン標準である OpenLineage と統合されたシステムからリネージ情報を取り込むことができます。ProcessOpenLineageRunEvent メソッドを使用して OpenLineage 形式のイベントを Data Lineage API に送信すると、Data Lineage API は OpenLineage メッセージの属性を Data Lineage API の対応する属性にマッピングします。

このドキュメントでは、これらのマッピングの参照テーブルを示します。

属性のマッピング

ProcessOpenLineageRunEvent REST API メソッドは、次のように OpenLineage 属性を Data Lineage API 属性にマッピングします。

Data Lineage API 属性 OpenLineage 属性
Process.name projects/PROJECT_NUMBER/locations/LOCATION/processes/HASH_OF_NAMESPACE_AND_NAME
Process.displayName Job.namespace + ":" + Job.name
Process.attributes Job.facets(保存データを参照)
Run.name projects/PROJECT_NUMBER/locations/LOCATION/processes/HASH_OF_NAMESPACE_AND_NAME/runs/HASH_OF_RUNID
Run.displayName Run.runId
Run.attributes Run.facets(保存データを参照)
Run.startTime eventTime
Run.endTime eventTime
Run.state eventType
LineageEvent.name projects/PROJECT_NUMBER/locations/LOCATION/processes/HASH_OF_NAMESPACE_AND_NAME/runs/HASH_OF_RUNID/lineageEvents/HASH_OF_JOB_RUN_INPUT_OUTPUTS_OF_EVENT(例: projects/11111111/locations/us/processes/1234/runs/4321/lineageEvents/111-222-333)
LineageEvent.EventLinks.source 入力(fqn は名前空間と名前を連結したもの)
LineageEvent.EventLinks.target 出力(fqn は名前空間と名前を連結したもの)
LineageEvent.startTime eventTime
LineageEvent.endTime eventTime
requestId メソッド ユーザーによって定義される

列レベルのリネージは、Managed Service for Apache Spark によって自動的に生成される場合にサポートされます。ただし、outputs オブジェクト内で columnLineage ファセットを使用している必要があります。このファセットを含むカスタム OpenLineage イベントは、テーブルレベルのリネージのみをマテリアライズし、 Google Cloud コンソールに列レベルのリネージをレンダリングしません。Managed Service for Apache Spark によって生成されるファセットの例を次に示します。

"outputs": [ {
  "namespace": "bigquery",
  "name": "project.dataset.outputtable",
  "columnLineage": {
      "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.39.0/integration/spark",
      "_schemaURL": "https://openlineage.io/spec/facets/1-2-0/ColumnLineageDatasetFacet.json#/$defs/ColumnLineageDatasetFacet",
      "fields": {
        "output_field": { // This is the name of the output field
          "inputFields": [
            {
              "namespace": "bigquery",
              "name": "project.dataset.inputtable",
              "field": "input_field", // This is the name of the input field
              "transformations": [
                {
                  "type": "DIRECT",
                  "subtype": "IDENTITY",
                  "description": "",
                  "masking": false
                }
              ]
            }
          ]
        }
      },
  }
}]

この例では、input_field と output_field の間に列レベルのリネージリンクを作成します。transformations フィールドを含める必要があります。このフィールドがないと、列レベルのリネージは取り込まれません。このファセットの Open Lineage 定義の詳細については、列レベルのデータセット ファセットをご覧ください。

FQN マッピング

次の表に、さまざまなシステムの OpenLineage 名前空間と名前のペアの例と、それに対応する Knowledge Catalog(旧称 Dataplex Universal Catalog)の完全修飾名(FQN)を示します。

システム OpenLineage 名前空間 OpenLineage 名 Knowledge Catalog の FQN
Athena awsathena://athena.{region_name}.amazonaws.com
  • {catalog}
  • {catalog}.{database}
  • {catalog}.{database}.{table}
  • athena:{catalogId}.{region}
  • athena:{catalogId}.{region}.{databaseId}
  • athena:{catalogId}.{region}.{databaseId}.{tableId}
AWS Glue arn:aws:glue:{region}:{account id} table/{database name}/{table name} aws_glue:table:{region}.{account id}.{database name}.{table name}
Azure Cosmos DB azurecosmos://{host}/dbs/{database} colls/{table}
  • cosmos-db:{host}.{database}
  • cosmos-db:{host}.{database}.{table}
Azure Data Explorer azurekusto://{host}.kusto.windows.net {database}/{table}
  • kusto:{host}.{region}.{database}
  • kusto:{host}.{region}.{database}.{table}
Azure Synapse sqlserver://{host}:{port}
  • {database}
  • {database}.{schema}
  • {database}.{schema}.{table}
非対応
BigQuery bigquery
  • {project id}.{dataset name}
  • {project id}.{dataset name}.{table name}
  • bigquery:{projectId}.{datasetId}
  • bigquery:{projectId}.{datasetId}.{assetId}
Cassandra cassandra://{host}:{port}
  • {keyspace}
  • {keyspace}.{table}
  • cassandra:{hostWithPort}.{keyspaceId}
  • cassandra:{hostWithPort}.{keyspaceId}.{tableId}
MySQL mysql://{host}:{port}
  • {database}
  • {database}.{table}
  • mysql:{hostWithPort}.{databaseId}
  • mysql:{hostWithPort}.{databaseId}.{tableId}
CrateDB crate://{host}:{port} {database}.{schema}.{table} 非対応
DB2 db2://{host}:{port}
  • {database}
  • {database}.{schema}
  • {database}.{schema}.{table}
  • db2:{dns}.{databaseId}
  • db2:{dns}.{databaseId}.{schemaId}
  • db2:{dns}.{databaseId}.{schemaId}.{tableId}
Hive hive://{host}:{port} {database}.{table} 非対応
MSSQL mssql://{host}:{port} {database}.{schema}.{table} 非対応
OceanBase oceanbase://{host}:{port} {database}.{table} 非対応
Oracle oracle://{host}:{port} {serviceName}.{schema}.{table} or {sid}.{schema}.{table}
  • oracle:{hostWithPort}.{databaseId}
  • oracle:{hostWithPort}.{databaseId}.{schemaId}
  • oracle:{hostWithPort}.{databaseId}.{schemaId}.{tableId}
Postgres postgres://{host}:{port}
  • {database}
  • {database}.{schema}
  • {database}.{schema}.{table}
  • postgresql:{hostWithPort}.{databaseId}
  • postgresql:{hostWithPort}.{databaseId}.{schemaId}
  • postgresql:{hostWithPort}.{databaseId}.{schemaId}.{tableId}
Teradata teradata://{host}:{port} {database}.{table} 非対応
Redshift redshift://{cluster_identifier}.{region_name}:{port}
  • {database}
  • {database}.{schema}
  • {database}.{schema}.{table}
  • redshift:{clusterId}.{region}.{port}.{databaseId}
  • redshift:{clusterId}.{region}.{port}.{databaseId}.{schemaId}
  • redshift:{clusterId}.{region}.{port}.{databaseId}.{schemaId}.{tableId}
Snowflake snowflake://{organization name}-{account name} or snowflake://{account-locator}(.{compliance})(.{cloud_region_id})(.{cloud})
  • {database}
  • {database}.{schema}
  • {database}.{schema}.{table}
  • snowflake:{accountName}.{databaseId}
  • snowflake:{accountName}.{databaseId}.{schemaId}
  • snowflake:{accountName}.{databaseId}.{schemaId}.{tableId}
Spanner spanner://{projectId}:{instanceId} {database}.{schema}.{table} Knowledge Catalog でサポートされていますが、データリネージではサポートされていません
Trino trino://{host}:{port}
  • {catalog}
  • {catalog}.{schema}
  • {catalog}.{schema}.{table}
  • trino:{hostWithPort}.{catalogId}
  • trino:{hostWithPort}.{catalogId}.{schemaId}
  • trino:{hostWithPort}.{catalogId}.{schemaId}.{tableId}
ABFSS(Azure Data Lake Gen2) abfss://{container name}@{service name}.dfs.core.windows.net {path}
  • abs:{serviceName}.{containerName}
  • abs:{serviceName}.{containerName}.{path}
DBFS(Databricks ファイル システム) dbfs://{workspace name} {path}
  • dbfs:{workspace}
  • dbfs:{workspace}.{path}
Cloud Storage gs://{bucket name} {object key}
  • gcs:{bucketName}
  • gcs:{bucketName}.{virtualPath}
HDFS hdfs://{namenode host}:{namenode port} {path}
  • hdfs:{namenodeHostWithPort}
  • hdfs:{namenodeHostWithPort}.{path}
Kafka kafka://{bootstrap server host}:{port} {topic} kafka:{serverHostWithPort}.{topicId}
ローカル ファイル システム file {path} filesystem:localhost.{path}
リモート ファイル システム file://{host} {path} filesystem:{hostWithPort}.{path}
S3 s3://{bucket name} {object key}
  • s3:{bucketName}
  • s3:{bucketName}.{objectKey}

名前空間プレフィックス s3a と s3n も受け入れられ、s3 に変換されます。
WASBS(Azure Blob Storage) wasbs://{container name}@{service name}.dfs.core.windows.net {object key}
  • abs:{serviceName}.{containerName}
  • abs:{serviceName}.{containerName}.{objectKey}
Pub/Sub トピック pubsub topic:{projectId}:{topicId} pubsub:topic:{projectId}.{topicId}
Pub/Sub サブスクリプション pubsub subscription:{projectId}:{subscriptionId} pubsub:subscription:{projectId}.{subscriptionId}

使用可能なその他の形式

OpenLineage では、次のシステムの標準の namespace/name ペアは定義されていませんが、Data Lineage API は、次の表に示す形式でリネージ イベントを受け入れます。Namespace custom の OpenLineage メッセージで参照されるリソースは、カスタムの完全修飾名として解釈されます。

システム OpenLineage 名前空間 OpenLineage 名 Knowledge Catalog の FQN
カスタム FQN custom {some reference} custom:{someReference}
Dataproc Metastore dataproc_metastore
  • dataproc_metastore:{projectId}.{location}.{instanceId}
  • dataproc_metastore:{projectId}.{location}.{instanceId}.{databaseId}
  • dataproc_metastore:{projectId}.{location}.{instanceId}.{databaseId}.{tableId}
  • dataproc_metastore:{projectId}.{location}.{instanceId}
  • dataproc_metastore:{projectId}.{location}.{instanceId}.{databaseId}
  • dataproc_metastore:{projectId}.{location}.{instanceId}.{databaseId}.{tableId}

次のステップ