OpenLineage を使用したカスタムジョブ

ドキュメント > Data Observability の概要 > Data Observability: Jobs Monitoring > OpenLineage を使用したカスタムジョブ

OpenLineage を使用したカスタムジョブはプレビュー版です。

概要

カスタムジョブは OpenLineage 標準を使用して、ジョブおよびリネージイベントを Datadog に送信します。カスタムジョブを使用すると、次のことができます。

失敗したジョブや長時間実行されているジョブを検出する
失敗したジョブや長時間実行されているジョブの根本原因を特定して解決する
データリネージを使用して、アップストリームの依存関係とダウンストリームのデータコンシューマーを把握する

次のような場合にカスタムジョブを使用します。

社内ツールやカスタム ETL スクリプトなど、Datadog がネイティブに統合していないシステムからリネージをキャプチャする
ネイティブの Datadog インテグレーションが利用できないジョブやオーケストレーターのリネージイベントを出力する

注: 構成を一元化し、すべてのアプリケーションに API キーを配布することを避けるために、Datadog Agent を OpenLineage プロキシとして設定できます。

前提条件

Datadog API キー。API キーとアプリケーションキーを参照してください。
お使いの Datadog のサイト URL。このページの例では datadoghq.com を使用しています。ホスト名をサイトのインテークエンドポイントに置き換えます。

ステップ 1: `START` イベントを送信する

OpenLineage イベントを Datadog に送信するには、次のいずれかのオプションを使用します。

注: Datadog が実行イベントを処理するには、jobType Job ファセットが必要です。

ジョブとそのデータセット間のリネージエッジも確認するには、イベントに inputs と outputs を含めます。データセットの名前空間は、各プラットフォームで Datadog が想定する形式と一致している必要があります。データセットの命名規則を参照してください。

生の OpenLineage RunEvent を JSON として Datadog のインテークエンドポイントに送信します。

curl -X POST "https://data-obs-intake.datadoghq.com/api/v1/lineage" \
  -H "Authorization: Bearer <DD_API_KEY>" \
  -H "Content-Type: application/json" \
  -d '{
        "eventTime": "2024-01-01T10:00:00Z",
        "eventType": "START",
        "run": { "runId": "<RUN_UUID>" },
        "job": {
          "namespace": "<YOUR_NAMESPACE>",
          "name": "<YOUR_JOB_NAME>",
          "facets": {
            "jobType": {
              "_producer": "<YOUR_PRODUCER_ID>",
              "_schemaURL": "https://openlineage.io/spec/facets/2-0-3/JobTypeJobFacet.json",
              "processingType": "BATCH",
              "integration": "custom",
              "jobType": "JOB"
            }
          }
        },
        "inputs": [
          {
            "namespace": "postgres://demo-db.example.com:5432",
            "name": "orders.public.orders"
          }
        ],
        "outputs": [
          {
            "namespace": "snowflake://myorg-myaccount",
            "name": "ANALYTICS.PUBLIC.ORDERS"
          }
        ],
        "producer": "<YOUR_PRODUCER_ID>"
      }'

手動で指定した HTTP トランスポートで OpenLineage Python クライアントを使用します。

from datetime import datetime
import uuid
from openlineage.client import OpenLineageClient, OpenLineageClientOptions
from openlineage.client.event_v2 import RunEvent, RunState, Job, Run, InputDataset, OutputDataset
from openlineage.client.facet_v2 import job_type_job

client = OpenLineageClient(
    url="https://data-obs-intake.datadoghq.com",
    options=OpenLineageClientOptions(api_key="<DD_API_KEY>")
)

event = RunEvent(
    eventType=RunState.START,
    eventTime=datetime.utcnow().isoformat(),
    run=Run(runId=str(uuid.uuid4())),
    job=Job(
        namespace="<YOUR_NAMESPACE>",
        name="<YOUR_JOB_NAME>",
        facets={
            "jobType": job_type_job.JobTypeJobFacet(
                processingType="BATCH",
                integration="custom",
                jobType="JOB"
            )
        }
    ),
    inputs=[
        InputDataset(
            namespace="postgres://demo-db.example.com:5432",
            name="orders.public.orders"
        )
    ],
    outputs=[
        OutputDataset(
            namespace="snowflake://myorg-myaccount",
            name="ANALYTICS.PUBLIC.ORDERS"
        )
    ],
    producer="<YOUR_PRODUCER_ID>"
)

client.emit(event)

OpenLineage 1.37.0 以降では、自動構成と最適化されたイベント配信のために Datadog トランスポートを使用します。

from datetime import datetime
import uuid
from openlineage.client import OpenLineageClient
from openlineage.client.event_v2 import RunEvent, RunState, Job, Run, InputDataset, OutputDataset
from openlineage.client.facet_v2 import job_type_job
from openlineage.client.transport.datadog import DatadogConfig, DatadogTransport

config = DatadogConfig(
    apiKey="<DD_API_KEY>",
    site="datadoghq.com"  # Change if using a different Datadog site
)

client = OpenLineageClient(transport=DatadogTransport(config))

event = RunEvent(
    eventType=RunState.START,
    eventTime=datetime.utcnow().isoformat(),
    run=Run(runId=str(uuid.uuid4())),
    job=Job(
        namespace="<YOUR_NAMESPACE>",
        name="<YOUR_JOB_NAME>",
        facets={
            "jobType": job_type_job.JobTypeJobFacet(
                processingType="BATCH",
                integration="custom",
                jobType="JOB"
            )
        }
    ),
    inputs=[
        InputDataset(
            namespace="postgres://demo-db.example.com:5432",
            name="orders.public.orders"
        )
    ],
    outputs=[
        OutputDataset(
            namespace="snowflake://myorg-myaccount",
            name="ANALYTICS.PUBLIC.ORDERS"
        )
    ],
    producer="<YOUR_PRODUCER_ID>"
)

client.emit(event)

DatadogConfig の代わりに環境変数を使用して Datadog トランスポートを構成することもできます。

export DD_API_KEY=<DD_API_KEY>
export DD_SITE=datadoghq.com
export OPENLINEAGE__TRANSPORT__TYPE=datadog

client = OpenLineageClient.from_environment()

ステップ 2: `RUNNING` イベントを送信する (オプション)

注: このステップはオプションです。RUNNING イベントを使用すると、ジョブが完了する前にそのステータスを確認できます。ジョブの完了ステータスのみをキャプチャする必要がある場合は、ステップ 3 に進みます。

ジョブの実行中に RUNNING イベントを送信して、Datadog の Jobs Monitoring で追跡します。runId は、START イベントで使用したものと同じものを使用します。

curl -X POST "https://data-obs-intake.datadoghq.com/api/v1/lineage" \
  -H "Authorization: Bearer <DD_API_KEY>" \
  -H "Content-Type: application/json" \
  -d '{
        "eventTime": "2024-01-01T10:02:00Z",
        "eventType": "RUNNING",
        "run": { "runId": "<RUN_UUID>" },
        "job": {
          "namespace": "<YOUR_NAMESPACE>",
          "name": "<YOUR_JOB_NAME>",
          "facets": {
            "jobType": {
              "_producer": "<YOUR_PRODUCER_ID>",
              "_schemaURL": "https://openlineage.io/spec/facets/2-0-3/JobTypeJobFacet.json",
              "processingType": "BATCH",
              "integration": "custom",
              "jobType": "JOB"
            }
          }
        },
        "producer": "<YOUR_PRODUCER_ID>"
      }'

from datetime import datetime
from openlineage.client.event_v2 import RunEvent, RunState, Job, Run
from openlineage.client.facet_v2 import job_type_job

running_event = RunEvent(
    eventType=RunState.RUNNING,
    eventTime=datetime.utcnow().isoformat(),
    run=Run(runId="<RUN_UUID>"),  # same runId as START
    job=Job(
        namespace="<YOUR_NAMESPACE>",
        name="<YOUR_JOB_NAME>",
        facets={
            "jobType": job_type_job.JobTypeJobFacet(
                processingType="BATCH",
                integration="custom",
                jobType="JOB"
            )
        }
    ),
    producer="<YOUR_PRODUCER_ID>"
)

client.emit(running_event)

from datetime import datetime
from openlineage.client.event_v2 import RunEvent, RunState, Job, Run
from openlineage.client.facet_v2 import job_type_job

running_event = RunEvent(
    eventType=RunState.RUNNING,
    eventTime=datetime.utcnow().isoformat(),
    run=Run(runId="<RUN_UUID>"),  # same runId as START
    job=Job(
        namespace="<YOUR_NAMESPACE>",
        name="<YOUR_JOB_NAME>",
        facets={
            "jobType": job_type_job.JobTypeJobFacet(
                processingType="BATCH",
                integration="custom",
                jobType="JOB"
            )
        }
    ),
    producer="<YOUR_PRODUCER_ID>"
)

client.emit(running_event)

ステップ 3: `COMPLETE` または `FAIL` イベントを送信する

ジョブが完了したら、START イベントで使用した runId と同じものを使用して、COMPLETE または FAIL イベントを送信します。

Success

curl -X POST "https://data-obs-intake.datadoghq.com/api/v1/lineage" \
  -H "Authorization: Bearer <DD_API_KEY>" \
  -H "Content-Type: application/json" \
  -d '{
        "eventTime": "2024-01-01T10:05:00Z",
        "eventType": "COMPLETE",
        "run": { "runId": "<RUN_UUID>" },
        "job": {
          "namespace": "<YOUR_NAMESPACE>",
          "name": "<YOUR_JOB_NAME>",
          "facets": {
            "jobType": {
              "_producer": "<YOUR_PRODUCER_ID>",
              "_schemaURL": "https://openlineage.io/spec/facets/2-0-3/JobTypeJobFacet.json",
              "processingType": "BATCH",
              "integration": "custom",
              "jobType": "JOB"
            }
          }
        },
        "producer": "<YOUR_PRODUCER_ID>"
      }'

Failure

curl -X POST "https://data-obs-intake.datadoghq.com/api/v1/lineage" \
  -H "Authorization: Bearer <DD_API_KEY>" \
  -H "Content-Type: application/json" \
  -d '{
        "eventTime": "2024-01-01T10:05:00Z",
        "eventType": "FAIL",
        "run": {
          "runId": "<RUN_UUID>",
          "facets": {
            "errorMessage": {
              "_producer": "<YOUR_PRODUCER_ID>",
              "_schemaURL": "https://openlineage.io/spec/facets/1-0-1/ErrorMessageRunFacet.json",
              "message": "Job failed: division by zero",
              "programmingLanguage": "Python",
              "stackTrace": "Traceback (most recent call last):\n  File \"job.py\", line 42, in run\n    result = total / count\nZeroDivisionError: division by zero"
            }
          }
        },
        "job": {
          "namespace": "<YOUR_NAMESPACE>",
          "name": "<YOUR_JOB_NAME>",
          "facets": {
            "jobType": {
              "_producer": "<YOUR_PRODUCER_ID>",
              "_schemaURL": "https://openlineage.io/spec/facets/2-0-3/JobTypeJobFacet.json",
              "processingType": "BATCH",
              "integration": "custom",
              "jobType": "JOB"
            }
          }
        },
        "producer": "<YOUR_PRODUCER_ID>"
      }'

Success

from datetime import datetime
from openlineage.client.event_v2 import RunEvent, RunState, Job, Run

complete_event = RunEvent(
    eventType=RunState.COMPLETE,
    eventTime=datetime.utcnow().isoformat(),
    run=Run(runId="<RUN_UUID>"),  # same runId as START
    job=Job(
        namespace="<YOUR_NAMESPACE>",
        name="<YOUR_JOB_NAME>",
        facets={
            "jobType": job_type_job.JobTypeJobFacet(
                processingType="BATCH",
                integration="custom",
                jobType="JOB"
            )
        }
    ),
    producer="<YOUR_PRODUCER_ID>"
)

client.emit(complete_event)

Failure

from datetime import datetime
from openlineage.client.event_v2 import RunEvent, RunState, Job, Run
from openlineage.client.facet_v2 import error_message_run

fail_event = RunEvent(
    eventType=RunState.FAIL,
    eventTime=datetime.utcnow().isoformat(),
    run=Run(
        runId="<RUN_UUID>",  # same runId as START
        facets={
            "errorMessage": error_message_run.ErrorMessageRunFacet(
                message="Job failed: division by zero",
                programmingLanguage="Python",
                stackTrace="Traceback (most recent call last):\n  File \"job.py\", line 42, in run\n    result = total / count\nZeroDivisionError: division by zero"
            )
        }
    ),
    job=Job(
        namespace="<YOUR_NAMESPACE>",
        name="<YOUR_JOB_NAME>",
        facets={
            "jobType": job_type_job.JobTypeJobFacet(
                processingType="BATCH",
                integration="custom",
                jobType="JOB"
            )
        }
    ),
    producer="<YOUR_PRODUCER_ID>"
)

client.emit(fail_event)

Success

from datetime import datetime
from openlineage.client.event_v2 import RunEvent, RunState, Job, Run
from openlineage.client.facet_v2 import job_type_job

complete_event = RunEvent(
    eventType=RunState.COMPLETE,
    eventTime=datetime.utcnow().isoformat(),
    run=Run(runId="<RUN_UUID>"),  # same runId as START
    job=Job(
        namespace="<YOUR_NAMESPACE>",
        name="<YOUR_JOB_NAME>",
        facets={
            "jobType": job_type_job.JobTypeJobFacet(
                processingType="BATCH",
                integration="custom",
                jobType="JOB"
            )
        }
    ),
    producer="<YOUR_PRODUCER_ID>"
)

client.emit(complete_event)

Failure

from datetime import datetime
from openlineage.client.event_v2 import RunEvent, RunState, Job, Run
from openlineage.client.facet_v2 import job_type_job, error_message_run

fail_event = RunEvent(
    eventType=RunState.FAIL,
    eventTime=datetime.utcnow().isoformat(),
    run=Run(
        runId="<RUN_UUID>",  # same runId as START
        facets={
            "errorMessage": error_message_run.ErrorMessageRunFacet(
                message="Job failed: division by zero",
                programmingLanguage="Python",
                stackTrace="Traceback (most recent call last):\n  File \"job.py\", line 42, in run\n    result = total / count\nZeroDivisionError: division by zero"
            )
        }
    ),
    job=Job(
        namespace="<YOUR_NAMESPACE>",
        name="<YOUR_JOB_NAME>",
        facets={
            "jobType": job_type_job.JobTypeJobFacet(
                processingType="BATCH",
                integration="custom",
                jobType="JOB"
            )
        }
    ),
    producer="<YOUR_PRODUCER_ID>"
)

client.emit(fail_event)

ステップ 4: Datadog で確認する

イベントを送信した後、以下のページを使用して、Datadog がイベントを受信して処理したことを確認します。

Jobs Overview: ジョブの実行が開始時刻、期間、ステータスとともに表示されることを確認します。
Lineage: ジョブ名を検索して、ジョブとイベントに含まれるデータセットとの関連を確認します。
OpenLineage events: イベントの検証ステータスを確認します: Ok、Warning、または Error。イベントを選択して Datadog が受信した正確なペイロードを確認するか、検証ステータスでフィルタリングして、正常に処理されなかったイベントのトラブルシューティングを行います。他の 2 つのビューで結果が期待どおりに表示されない場合は、ここから開始してください。

ログとジョブ実行の相関付け

アプリケーションログを Datadog のジョブ実行と相関させるには、@openlineage.run_id 属性に OpenLineage の実行 ID を含めてログを出力します。その値を、OpenLineage の実行イベントで送信するものと同じ runId に設定します。Datadog はこの属性を使用して、ログを対応するジョブ実行に関連付けます。

実行 ID をどのように付与するかは、ログの設定によって異なります。カスタム属性を含むログを Datadog に送信する方法の詳細については、ログの収集とインテグレーションを参照してください。

データセットの命名規則

カスタムジョブのリネージを Datadog のネイティブ Integrations ですでに追跡されているデータセットに接続するには、Datadog がそのプラットフォームに対して想定している正確な namespace と name を使用して、イベントに inputs と outputs を含めます。たとえば、カスタムジョブの outputs で Snowflake テーブルを正しい名前空間と名前で参照すると、リネージグラフ内の既存のデータセットノードにリンクされます。

Datadog は、データセットをアカウント、データベース、スキーマ、テーブルの階層に解決します。名前の構成要素が想定より少ない場合 (例: database.table ではなく database.schema.table)、Datadog はリネージグラフ内の最も近い上位ノードにフォールバックします。

プラットフォーム	名前空間	名前
BigQuery	`bigquery`	`{project}.{dataset}.{table}`
Snowflake	`snowflake://{org}-{account}`	`{database}.{schema}.{table}`
Redshift	`redshift://{aws_account_id}:{region}:{cluster}`	`{database}.{schema}.{table}`
PostgreSQL	`postgres://{host}:{port}`	`{database}.{schema}.{table}`
Databricks	`databricks://{workspace-url}`	`{database}.{schema}.{table}`
Trino	`trino://{host}:{port}`	`{catalog}.{schema}.{table}`
AWS Glue	`arn:aws:glue:{region}:{accountId}`	`{database}.{table}`
S3	`s3://{bucket}`	`{path}`

ここに記載されていないプラットフォームについては、OpenLineage の命名規則に従ってください。

次の例は、PostgreSQL テーブルから読み取り、Snowflake テーブルに書き込むジョブを示しています。

"inputs": [
  {
    "namespace": "postgres://db.example.com:5432",
    "name": "mydb.public.raw_orders"
  }
],
"outputs": [
  {
    "namespace": "snowflake://myorg-myaccount",
    "name": "ANALYTICS.PUBLIC.ORDERS"
  }
]

注: データセットの名前空間が認識されない場合、Datadog は引き続きその名前空間のリネージノードを作成しますが、Data Observability 製品には表示しません。データセットをカタログおよびリネージグラフに表示させるには、認識される名前空間形式を使用してください。

ファセットは、OpenLineage イベントに付加される構造化されたメタデータです。各ファセットには、_producer (それを生成したシステムを識別する URI) と _schemaURL (その JSON スキーマを参照する URI) が必要です。

`JobTypeJobFacet`

jobType ジョブファセットは必須です。これは、Datadog がジョブをどのように分類し表示するかを決定します。

`integration`値

ネイティブインテグレーションでまだサポートされていないテクノロジーで実行されるジョブについては、ネイティブインテグレーション用に予約されていない任意の値 (例: my-pipeline) を使用してください。この値は、ジョブの種類を示すために Datadog 全体で使用されます。以下の予約済みの値は、Datadog のネイティブインテグレーションで使用されます。カスタムジョブに予約済みの値を使用すると、予期しない動作が発生する可能性があり、サポートされていません。

値	プラットフォーム
`<YOUR_VALUE>`	カスタムプラットフォームまたはサポートされていないプラットフォーム
`SPARK`	Apache Spark (ネイティブインテグレーションのみ。カスタムジョブには使用しないでください)
`AIRFLOW`	Apache Airflow
`DBT`	dbt
`BIGQUERY`	Google BigQuery
`SNOWFLAKE`	Snowflake
`TRINO`	Trino
`ICEBERG`	Apache Iceberg
`TABLEAU`	Tableau

`processingType` 値

BATCH または STREAMING。

`jobType`値

一般的な値には、JOB、TASK、DAG、MODEL、COMMAND、および QUERY が含まれます。

注: jobType が QUERY に設定されている場合、Datadog はジョブのリネージノードを生成しません。

ファセット	Datadog の動作
`parent`	リネージグラフに親子関係のジョブ階層を作成します
`errorMessage`	`error.message` タグと `error.stack` タグを含むエラースパンを生成します
`tags`	ジョブまたは実行にカスタムタグを追加します。`_dd.ol_service` の値は Datadog サービス名にマッピングされます
`sql`	SQL クエリを解析およびマスクし、クエリイベントを生成します

注: tagsファセット内の各タグには、key、value、および source を含める必要があります。

カスタム OpenLineage タグを Data Job Monitors で利用できるようにするには、source を正確に USER に設定してください。source が欠落しているか異なるタグは、Data Job Monitors ではカスタムタグとして利用できません。

job.facets.tags と run.facets.tags のタグは動作が異なります。

Job facet tags: 基盤となるジョブトレースにタグとして追加されるため、Jobs Overview ページや Trace Explorer でそれらを使用してフィルタリングできます。source が USER の場合、それらは Data Job Monitors でも利用できます。team や owner など、安定したジョブプロパティにはジョブファセットタグを使用してください。
Run facet tags: source が USER の場合、Data Job Monitors で利用できます。これらは、Jobs Overview ページや Trace Explorer で検索やフィルタリングを行うための個別のタグとしては追加されません。実行ごとに値が異なる可能性がある場合は、実行ファセットタグを使用してください。