mcpField
D
data-engineering

data-engineering

github.com/travisjneumanデータ・データベース106100 回インストール今日更新
リポジトリ

data-engineering について

データパイプライン、変換、ストリーム処理、ウェアハウスのスキーマ、データ品質チェックを設計するデータエンジニアリングスキルです。Airflow、Dagster、dbt、Kafka、Great Expectations などのツールに関するパターンと例を紹介します。

主な機能

  1. 01ETL と ELT のアーキテクチャパターン
  2. 02Airflow と Dagster によるオーケストレーション
  3. 03dbt の変換例
  4. 04Kafka のストリーム処理
  5. 05ウェアハウス設計と品質管理
  6. 06増分処理と冪等なパイプライン

活用シーン

  1. 01ETL と ELT の方式選択
  2. 02オーケストレーションされたデータパイプラインの構成
  3. 03ウェアハウステーブルと品質チェックの設計
  4. 04増分処理とバックフィルの計画

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/travisjneuman/.claude --skill data-engineering

インストール後、Claude Code で/data-engineeringを入力して読み込まれたか確認します。

詳細

データエンジニアリング

パイプラインアーキテクチャ

ETL と ELT
パターン使用する場面ツール
ETL読み込み前に変換する場合、データ品質が重要な場合Airflow + custom、Spark
ELT生データ → データウェアハウス → インプレース変換Fivetran + dbt、Airbyte + dbt
オーケストレーション

Apache Airflow:

from airflow.decorators import dag, task
from datetime import datetime

@dag(schedule="@daily", start_date=datetime(2024, 1, 1), catchup=False)
def my_pipeline():
    @task()
    def extract() -> dict:
        return {"data": "extracted"}

    @task()
    def transform(data: dict) -> dict:
        return {"transformed": True}

    @task()
    def load(data: dict):
        # Load to warehouse
        pass

    raw = extract()
    transformed = transform(raw)
    load(transformed)

my_pipeline()

Dagster（新規プロジェクトにおすすめ）：

from dagster import asset, Definitions

@asset
def raw_users():
    return extract_from_source()

@asset
def cleaned_users(raw_users):
    return clean_and_validate(raw_users)

dbt による変換

-- models/marts/dim_customers.sql
{{ config(materialized='table', schema='marts') }}

WITH source AS (
    SELECT * FROM {{ ref('stg_customers') }}
),
orders AS (
    SELECT customer_id, COUNT(*) as order_count, SUM(amount) as total_spent
    FROM {{ ref('stg_orders') }}
    GROUP BY customer_id
)
SELECT
    s.customer_id,
    s.name,
    s.email,
    COALESCE(o.order_count, 0) as lifetime_orders,
    COALESCE(o.total_spent, 0) as lifetime_value
FROM source s
LEFT JOIN orders o ON s.customer_id = o.customer_id

ストリーム処理

Apache Kafka:

from confluent_kafka import Producer, Consumer

# Producer
producer = Producer({'bootstrap.servers': 'localhost:9092'})
producer.produce('events', key='user_123', value=json.dumps(event))
producer.flush()

# Consumer
consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'auto.offset.reset': 'earliest'
})
consumer.subscribe(['events'])

データウェアハウスのスキーマ設計

スタースキーマ
  • ファクトテーブル： 測定可能なイベント（注文、クリック、取引）
  • ディメンションテーブル： 説明的なコンテキスト（顧客、商品、日付）
  • 緩やかに変化するディメンション： タイプ 1（上書き）、タイプ 2（バージョン管理された行）、タイプ 3（以前の列）
データ品質
  • Great Expectations： スキーマ検証、統計テスト、カスタムの期待値
  • dbt テスト： not_null、unique、accepted_values、relationships、カスタム SQL テスト
  • データコントラクト： スキーマ進化のポリシー、後方互換性の要件

主なパターン

  • 冪等性のあるパイプライン： 同じ入力からは常に同じ出力が生成され、安全に再実行できます
  • インクリメンタルモデル： 新規または変更されたデータのみを処理し、updated_at のウォーターマークを使用します
  • デッドレターキュー： パイプラインをブロックせずに、失敗したレコードを調査用に振り分けます
  • バックフィル戦略： 時間でパーティション分割されたテーブルにより、対象を絞った過去データの再処理が可能になります

コメント

Markdown 対応