データエンジニアリングにおけるデータのパイプラインを構成できるビルダーパターンを活用

アナログ: カスタムピザを注文する

カスタムピザを注文するようなビルダーパターンを考えてください。 残酷、ソース、チーズ、そして1回ずつトッピングする。 ピザビルダー(シェフ)は、これらの成分を完成したピザに結合する方法を知っています。 同じビルダーは、マーゲリータ、ハワイアン、または肉愛好家のパイを生成できます。 同様に、データパイプラインビルダーは、同じセットのビルダーメソッドからソース、変換、シンクの異なる組み合わせを組み立てることができます。

なぜデータパイプラインは、構成可能な設計が必要

データのパイプラインは、ほとんど静的です。S3 バケットから CSV ファイルを摂取し、データを倉庫に読み込みるパイプラインは、JSON、ストリーミング ソース、または追加の強化手順をサポートする迅速な場合があります。構成可能な設計がなければ、そのような変更を加えることは、重複やエラーのためのレシピ - 大量のコードをコピーおよび変更することを意味しています。

• [] ソースシステム:[]] バッチファイルからイベントストリームへのシフト、またはデータベースコネクタの切り替え。
• 進化する変換:]]]データのクリーンアップ、機能工学の追加、または新しい参照テーブルとの結合。
• []複数の宛先:[]]]] 複数のデータストア(例えば、BigQuery、Snowflake、およびリアルタイムダッシュボード)に結果を同じパイプラインに書きます。
• [] バリアントをテスト・ステージング:[[ 開発と生産データに対して、コード変更なしで同一のロジックを実行します。

ビルダーパターンは、エンジニアを宣言して、これらのニーズに直接対処します。 パイプラインは、宣言的に] - コンポーネントが含まれているかを定義し、それらがどのように接続するかを定義します。 これにより、基礎的なアセンブリロジックは変更されません。

構成可能なデータパイプラインのコアコンポーネント

ビルダーパターンを適用するには、データパイプラインは、ディスクリート、複合構造ブロックに分割する必要があります。

データソース

パイプラインは、ファイルシステム、データベース、ストリーミングプラットフォーム(Kafka)、API、データ湖などのソースから始まります。各ソースには、独自の構成(パス、認証情報、スキーマ、ポーリング間隔)があります。ビルダーは、、、またはなどのメソッドを供給することができます。

変革のステップ

変換はデータを操作したり、データを豊かにしたりします。一般的な例には、フィルタリング行、ネストされたJSONを解析し、メトリックを集計し、データセットに加わる。ビルダーメソッドは、、および[]など、このメソッドは、エンジニアが変換を流暢にシーケンスできるようにします。

データシンク

シンクは、リレーショナルデータベース、クラウドストレージ、メッセージキュー、または分析エンジンの処理データランドの処理場所です。ビルダーは、複数のシンクをとでサポートし、チェーンを複数の宛先に送信することもできます。

コネクターおよびミドルウェア

ソースとシンクを超えて、パイプラインは、多くの場合、エラーハンドラ、レートリミッター、スキーマバリデータ、およびホックの監視を必要とします。 これらのクロスカットの懸念は、または[のようなビルダーの手順として簡単に追加されます。

パイプラインのためのビルダーパターンの実装

典型的な実装には、構成オプションを収集する [] パイプラインビルダークラス] と、完全に構築されたパイプラインオブジェクトを検証し、返す [] メソッド が含まれます。 ビルダーは、ビルド自体をチェーンするために戻すフルエントなメソッドを公開します。

class PipelineBuilder:
 def __init__(self):
 self._source = None
 self._transformations = []
 self._sinks = []
 self._retry_policy = None

 def with_source(self, source):
 self._source = source
 return self

 def add_transform(self, transform):
 self._transformations.append(transform)
 return self

 def add_sink(self, sink):
 self._sinks.append(sink)
 return self

 def with_retry(self, retry_policy):
 self._retry_policy = retry_policy
 return self

 def build(self):
 if not self._source or not self._sinks:
 raise ValueError("Source and at least one sink are required")
 return Pipeline(self._source, self._transformations, self._sinks, self._retry_policy)

ビルダーを使用して、パイプラインの作成は宣言されます。

pipeline = (PipelineBuilder()
 .with_source(S3CsvSource(bucket="data-landing", prefix="orders/"))
 .add_transform(FilterTransform(condition="status == 'active'"))
 .add_transform(AggregateTransform(group_by="customer_id", metrics=["sum(amount)"]))
 .add_sink(DatabaseSink(connection="prod_db", table="customer_orders"))
 .add_sink(ParquetSink(path="s3://analytics/orders/"))
 .with_retry(RetryPolicy(max_attempts=3, backoff_seconds=5))
 .build())

このアプローチは、設定を集中化し、ステージングや生産環境の異なるパラメータで同じビルダーを再利用するのを容易にします。

実世界応用: 適用範囲が広いETLのパイプラインを造る

複数の地域から毎日注文データを摂取し、きれいにし、標準化し、カテゴリ別に毎日収益を計算し、レポートデータベースとデータ湖の両方に結果をロードする必要があるeコマース会社を考えてみましょう。 ビルダーパターンを使用して、再利用可能なOrderETLBuilderを作成します。

1. [] Define ソースコンフィグ:[] 各地域の注文は、異なるデータベース(PostgreSQL、MySQL)から来ていますが、共有CSV形式にエクスポートします。 ビルダーは ]を提供します。
2. []標準変換を追加します:[]]データクレンジング(nullオーダーIDを削除、通貨コードを検証)とエンリッチメント(カテゴリを取得するには製品カタログと同行)。 これらはと[を介して追加されます。
3. 集合集合: ]。
4. ]複数のシンクにルート: ]と。
5. []ビルドと実行:]]]と同じビルダーは、最初に、EU地域のみのテストを読み取り、その後、生産のためのすべての地域にスワップするパイプラインを構築することができます。

このパターンは、コード重複を劇的に減らします。この会社は、地域や環境ごとに複数のアドホックスクリプトではなく、一つのビルダークラスを維持します。

利点 要約

• []柔軟性:]] 実行ロジックに触れることなくパイプラインの動作を変更します。新しい変換を追加する必要がありますか? [を新しいステップで呼び出します。
• メンテナンス性:]] パイプライン定義は、高レベルのレシピのように読みます。各コンポーネントの構成は分離され、デバッグとコードレビューをストレートフォワードにしています。
• []再利用可能な:]] ビルダーは、ライブラリとしてパッケージ化できます。 チームは、プロジェクト全体で同じビルダーを再利用し、入力パラメータのみを調整します。
• :]]]] 新規コンポーネントタイプを追加(例えば、ストリーミングシンク) ビルダーを拡張するだけで、パイプラインアセンブリ全体を書き換える必要もありません。
• 試験性:]] ビルダーは、モックソースとシンクでテストパイプラインを作成でき、パイプラインアセンブリロジック自体の分離ユニットテストを可能にします。

データエンジニアリングにおけるビルダーパターンの使用に最適なプラクティス

ビルダーの純粋な構成を保って下さい

ビルダーは、構成を収集し検証するだけでなければなりません。実際のパイプラインの実行は、[]の責任である必要があります。 ]によって構成されるオブジェクト。この分離は、ビルダーをシンプルかつテスト可能に保ちます。