データパイプラインとは？データの移動と変換の解説

データパイプラインは、データが生成される場所から使用される場所へと移動させ、その過程で変換と検証を行う一連の自動化されたステップです。

データパイプラインはデータプラットフォームの配管です。データベース、API、イベントストリームといったソースから生データを取り込み、クリーンで有用な形へと変換し、アナリストやモデルが利用できる場所に配置します。CI/CDのpipelineと同じように、順番に実行される個別の自動化されたステージから構成されます。

データパイプラインとは何か

データパイプラインは、ソースからデスティネーションへとデータを処理する、順序付けられたステップ - 抽出、クリーニング、変換、検証、ロード - の集合です。各ステップには入力と出力があり、pipelineは再実行してその結果を再現できます。デスティネーションはしばしばdata warehouse、data lake、またはfeature storeです。

batch対ストリーミング

batchのpipelineは、スケジュールに従ってデータをまとまりで処理します。たとえば毎晩、前夜分を処理するといった具合です。ストリーミングのpipelineは、イベントが到着するのに合わせて低レイテンシで継続的に処理します。多くのプラットフォームは両方を実行します。重い履歴処理にはbatchを、新鮮でほぼリアルタイムのデータにはストリーミングを使います。

オーケストレーションと依存関係

実際のpipelineには互いに依存し合うステップがあるため、通常はDAG（有向非巡回グラフ）としてモデル化され、AirflowやDagsterのようなオーケストレーターによって実行されます。オーケストレーターは何を並列に実行できるかを判断し、失敗したステップをリトライし、どの実行が成功したかを追跡します。

CIでデータパイプラインをテストする

pipelineのコードはコードなので、CIに属します。変換にlintをかけ、小さなfixtureに対してユニットテストし、pull requestのたびにサンプルデータに対してpipelineをエンドツーエンドで実行します。壊れた変換がwarehouseを破損させる前に捕まえる方が、後から修正するよりはるかに安上がりです。

# CI for a data pipeline
steps:
  - run: pytest tests/transforms   # unit-test transforms
  - run: python run_pipeline.py --sample data/fixtures
  - run: great_expectations checkpoint run nightly

Latchkeyについて

pipelineのテストjobはサンプルデータセットを取得し、ローカルデータベースのようなservice containerを立ち上げます。Latchkeyでは、それらのfixtureや依存関係を実行間でcacheし、ソースのAPIやobject storeが一時的に不調になったときに一過性の失敗を自動リトライできます。これにより、不安定なデータ取得が、本来良好なbuildに赤いXを付けてしまうことがなくなります。

重要なポイント

• データパイプラインは、データをソースからデスティネーションへ移動・変換する、順序付けられた自動化ステップの集合です。
• pipelineはbatch（スケジュールされたまとまり）またはストリーミング（継続的なイベント）で実行され、複雑なものはDAGとしてモデル化されます。
• pipelineのコードは、本番データに触れる前に小さなfixtureに対してCIでテストされるべきです。

関連ガイド