OpenLineage

OpenLineage 是面向数据血缘采集的开放标准。本指南说明如何将 OpenLineage RunEvent、JobEvent 和 DatasetEvent 接入观测云，适用于 Airflow、Spark、dbt 以及直接使用 OpenLineage 客户端的应用。

前置条件¶

开始前，请准备：

• 当前站点的数据接入地址，例如 https://openway.guance.com；其他站点或私有部署请使用实际地址。
• 目标工作空间的 Token。请将 Token 作为密钥管理，不要写入源码或提交到版本库。
• 支持 OpenLineage HTTP transport 的数据任务或客户端。
• 从任务运行环境到数据接入地址的 HTTPS 网络连通性。

下文使用以下占位符：

接入协议¶

完整 URL 为：

<OPENLINEAGE_BASE_URL>/v1/write/openlineage

每次请求只发送一个事件，不支持 JSON 数组或 OpenLineage batch 接口。建议保持单条事件远小于上限，避免超大的 Facet 影响发送和重试。

认证¶

支持以下三种认证方式：

如果请求同时提供多种 Token，优先级为 query token、X-Token、Bearer。不要在一个请求中配置多个不同的 Token。

事件要求¶

事件必须符合 OpenLineage 规范。三种事件都必须包含 eventTime（ISO 8601 时间）、producer（生产者 URI）和 schemaURL（规范版本地址）；观测云 根据顶层的 run、job、dataset 识别类型：

RunEvent 和 JobEvent 有数据输入或输出时填写 inputs、outputs，每个 Dataset 应有 namespace 和 name。同一次运行的 RunEvent 必须使用相同的 runId、job.namespace 和 job.name；通常先发送 START，结束时再发送 COMPLETE、FAIL 或 ABORT。

事件分类与记录类型¶

接入后，source 表示原始 OpenLineage 事件的类型，lineage_record_type 表示 观测云 从该事件投影出的记录类型。它们不是同一个维度：同一个 RunEvent 可以生成多条记录，但这些记录的 source 都是 run_event。

event 保存经过敏感信息脱敏的完整事件 JSON（lineage_message）；job、dataset 是实体记录；dataset_edge 表示 Job/Run 的输入或输出 Dataset；schema_version 是 Schema 摘要；field_edge 是字段级血缘。无相关输入、输出或 Facet 时，不会生成对应的派生记录。无效事件的审计记录可能出现 source=unknown_event、lineage_record_type=ingest_error；这不是第四种可发送的事件类型。

source 是 观测云 保留的分类字段。如果原始 JSON 顶层另有字符串 source，投影记录会将其保存在 src_source，不会覆盖 观测云 的 source；原始事件的该字段仍可在 event 记录的 lineage_message 中查看。

查询字段清单¶

下表是 观测云 提取到一级字段的字段清单，并非 OpenLineage 原始 JSON 的全部字段。字段按事件类型、记录类型及 Facet 是否存在而出现；某个字段未出现，不等于其值为空。需要完整事件时查看 lineage_record_type=event 的 lineage_message（敏感值可能已脱敏）。同一事件的多条投影记录通过 event_id 关联；run_id、job_id、dataset_id 分别用于关联运行、Job 和 Dataset。

通用字段与事件原文¶

Run 与 Job 字段¶

身份字段会随同一事件的派生记录传递；Job 名称、状态及 Facet 提取字段主要位于 event，JobEvent 的 Job 元数据还位于 job 记录。

Dataset、Schema 与血缘边字段¶

dataset 记录只由 DatasetEvent 生成；dataset_edge 由 RunEvent 或 JobEvent 的输入、输出生成。schema_version 只保留 Schema 摘要，完整 Schema 仍在 lineage_message 的原始 Facet 中。

ingest_error 记录另有 ingest_error_reason_code（拒收原因）、ingest_error_payload_hash（载荷摘要）和 ingest_error_payload_size（载荷字节数）；为避免泄漏，不保存原始错误载荷。

快速验证¶

将以下内容保存为 event.json，并为每次测试生成新的 runId：

{
  "eventType": "START",
  "eventTime": "2026-10-21T08:00:00Z",
  "run": {
    "runId": "4a8f90a7-79c0-4f80-89e7-6b6c5d4fc79a"
  },
  "job": {
    "namespace": "example.scheduler",
    "name": "daily_orders"
  },
  "inputs": [
    {
      "namespace": "postgresql://warehouse.example:5432",
      "name": "sales.orders"
    }
  ],
  "outputs": [
    {
      "namespace": "postgresql://warehouse.example:5432",
      "name": "analytics.daily_orders"
    }
  ],
  "producer": "https://example.com/openlineage/integration",
  "schemaURL": "https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/RunEvent"
}

使用 Bearer Token 发送：

curl --fail-with-body --silent --show-error \
  -X POST '<OPENLINEAGE_BASE_URL>/v1/write/openlineage' \
  -H 'Authorization: Bearer <WORKSPACE_TOKEN>' \
  -H 'Content-Type: application/json' \
  --data-binary @event.json

使用 gzip 压缩发送：

gzip -c event.json | curl --fail-with-body --silent --show-error \
  -X POST '<OPENLINEAGE_BASE_URL>/v1/write/openlineage' \
  -H 'Authorization: Bearer <WORKSPACE_TOKEN>' \
  -H 'Content-Type: application/json' \
  -H 'Content-Encoding: gzip' \
  --data-binary @-

请求成功后，再使用相同的 runId 和 Job 标识发送 COMPLETE 事件，验证一次完整的运行生命周期。

客户端配置¶

OpenLineage Python Client¶

OpenLineage Python Client 的 api_key 认证会生成 Bearer Header。以下配置同时启用 gzip 和对临时错误的重试：

export OPENLINEAGE__TRANSPORT__TYPE=http
export OPENLINEAGE__TRANSPORT__URL='<OPENLINEAGE_BASE_URL>'
export OPENLINEAGE__TRANSPORT__ENDPOINT='/v1/write/openlineage'
export OPENLINEAGE__TRANSPORT__AUTH__TYPE=api_key
export OPENLINEAGE__TRANSPORT__AUTH__APIKEY='<WORKSPACE_TOKEN>'
export OPENLINEAGE__TRANSPORT__COMPRESSION=gzip
export OPENLINEAGE__TRANSPORT__RETRY='{"total":5,"connect":5,"read":5,"backoff_factor":0.3,"status_forcelist":[429,500,502,503,504],"allowed_methods":["POST"]}'

不同客户端版本支持的配置项可能不同，升级前请核对 Python Client 配置说明。

Apache Airflow¶

安装与当前 Airflow 版本兼容的 OpenLineage Provider，然后配置 HTTP transport：

pip install apache-airflow-providers-openlineage

export AIRFLOW__OPENLINEAGE__NAMESPACE='production-airflow'
export AIRFLOW__OPENLINEAGE__TRANSPORT='{"type":"http","url":"<OPENLINEAGE_BASE_URL>","endpoint":"/v1/write/openlineage","auth":{"type":"api_key","apiKey":"<WORKSPACE_TOKEN>"},"compression":"gzip"}'

Provider 与 Airflow 有版本兼容要求。安装或升级时请查看 Airflow OpenLineage Provider 文档。

Apache Spark¶

加载与 Spark 版本兼容的 OpenLineage Spark 集成包，并设置以下属性：

spark.extraListeners=io.openlineage.spark.agent.OpenLineageSparkListener
spark.openlineage.transport.type=http
spark.openlineage.transport.url=<OPENLINEAGE_BASE_URL>
spark.openlineage.transport.endpoint=/v1/write/openlineage
spark.openlineage.transport.auth.type=api_key
spark.openlineage.transport.auth.apiKey=<WORKSPACE_TOKEN>
spark.openlineage.transport.compression=gzip
spark.openlineage.namespace=production-spark

集成包的加载方式与版本兼容关系，请参考 OpenLineage Spark 安装文档。

dbt¶

安装 openlineage-dbt，复用 Python Client 的 transport 配置，并使用 dbt-ol 替代 dbt：

pip install openlineage-dbt

export OPENLINEAGE__TRANSPORT__TYPE=http
export OPENLINEAGE__TRANSPORT__URL='<OPENLINEAGE_BASE_URL>'
export OPENLINEAGE__TRANSPORT__ENDPOINT='/v1/write/openlineage'
export OPENLINEAGE__TRANSPORT__AUTH__TYPE=api_key
export OPENLINEAGE__TRANSPORT__AUTH__APIKEY='<WORKSPACE_TOKEN>'
export OPENLINEAGE__TRANSPORT__COMPRESSION=gzip
export OPENLINEAGE_NAMESPACE='production-dbt'

dbt-ol run

不要只设置 OPENLINEAGE_URL：客户端默认路径通常是 /api/v1/lineage，本接入要求显式配置 /v1/write/openlineage。更多运行模式和适配器支持情况见 OpenLineage dbt 集成文档。

响应与重试¶

发送方必须允许重复投递，并为同一次运行保持稳定的 runId。不要重试确定性的 4xx 错误，否则无效事件可能持续占用发送队列。

验证与排障¶

完成配置后，建议按以下顺序验证：

1. 使用 curl 发送最小 START 事件，确认返回 2xx。
2. 使用相同的 Run 和 Job 标识发送终态事件。
3. 在观测云中按 Job namespace、Job name 或 runId 查找对应记录。
4. 检查生产者日志，确认没有持续重试、认证错误或事件序列化错误。

常见问题：

安全建议¶

• 生产环境只使用 HTTPS，并保持证书校验开启。
• 优先使用 Bearer 或 X-Token，避免在 URL 中携带 Token。
• 通过 Secret、密钥管理服务或运行环境注入 Token；不要写入镜像、配置模板、命令历史或日志。
• 定期轮换 Token，并在泄漏后立即撤销旧 Token。
• Facet 可能包含 SQL、Schema 或业务元数据；发送前评估敏感信息，并按最小必要原则采集。

参考资料¶

• OpenLineage 核心规范
• OpenLineage 事件示例
• OpenLineage Python Client 配置
• Apache Airflow OpenLineage Provider
• OpenLineage Spark 集成
• OpenLineage dbt 集成