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 方面，则支持列级沿袭。包含此 facet 的自定义 OpenLineage 事件仅具体化表级沿袭，不会在 Google Cloud 控制台中呈现列级沿袭。以下是 Managed Service for Apache Spark 生成的 facet 示例：
"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 字段；否则，系统不会提取列级沿袭。如需详细了解此 facet 的 Open Lineage 定义，请参阅列级数据集 facet。
FQN 映射
下表提供了各种系统的 OpenLineage 命名空间和名称对示例，以及其等效的 Knowledge Catalog（原 Dataplex Universal Catalog）完全限定名称 (FQN)：
| 系统 | OpenLineage 命名空间 | OpenLineage 名称 | Knowledge Catalog FQN |
|---|---|---|---|
| Athena | awsathena://athena.{region_name}.amazonaws.com |
|
|
| 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} |
|
| Azure Data Explorer | azurekusto://{host}.kusto.windows.net |
{database}/{table} |
|
| Azure Synapse | sqlserver://{host}:{port} |
|
不支持 |
| BigQuery | bigquery |
|
|
| Cassandra | cassandra://{host}:{port} |
|
|
| MySQL | mysql://{host}:{port} |
|
|
| CrateDB | crate://{host}:{port} |
{database}.{schema}.{table} |
不支持 |
| DB2 | db2://{host}:{port} |
|
|
| 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} |
|
| Postgres | postgres://{host}:{port} |
|
|
| TeraData | teradata://{host}:{port} |
{database}.{table} |
不支持 |
| Redshift | redshift://{cluster_identifier}.{region_name}:{port} |
|
|
| Snowflake | snowflake://{organization name}-{account name} or snowflake://{account-locator}(.{compliance})(.{cloud_region_id})(.{cloud}) |
|
|
| Spanner | spanner://{projectId}:{instanceId} |
{database}.{schema}.{table} |
Knowledge Catalog 支持，但数据沿袭不支持 |
| Trino | trino://{host}:{port} |
|
|
| ABFSS (Azure Data Lake Gen2) | abfss://{container name}@{service name}.dfs.core.windows.net |
{path} |
|
| DBFS（Databricks 文件系统） | dbfs://{workspace name} |
{path} |
|
| Cloud Storage | gs://{bucket name} |
{object key} |
|
| HDFS | hdfs://{namenode host}:{namenode port} |
{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} |
命名空间前缀 s3a 和 s3n 也可接受，并会转换为 s3 |
| WASBS (Azure Blob Storage) | wasbs://{container name}@{service name}.dfs.core.windows.net |
{object key} |
|
| 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 接受这些系统的沿袭事件，前提是这些事件的格式符合下表中的说明。在 OpenLineage 消息中引用且命名空间为 custom 的资源会被解读为自定义的完全限定名称。
| 系统 | OpenLineage 命名空间 | OpenLineage 名称 | Knowledge Catalog FQN |
|---|---|---|---|
| 自定义 FQN | custom |
{some reference} |
custom:{someReference} |
| Dataproc Metastore | dataproc_metastore |
|
|
后续步骤
- 了解如何与 OpenLineage 集成。
- 请参阅完全限定名称参考文档。
- 探索 Data Lineage API。
- 了解如何查看沿袭信息。