etl-pipelines

etl-pipelines

目的

このスキルを使用すると、OpenClaw でデータエンジニアリングのワークフローにおけるデータの抽出、変換、ロードを行う ETL パイプラインを設計、実装できます。データベース、ファイル、API などの構造化データソースの処理に重点を置き、分析やレポート作成に向けた効率的なデータフローを実現します。

使用する場面

バッチ処理、リアルタイムデータ取り込み、データ移行用のデータパイプラインを構築する際に、このスキルを使用します。大規模なデータセット（例：1 TB 超）の処理、Apache Spark や AWS Glue などのツールとの統合、BI ダッシュボード向け ETL の自動化を行う場面に適用します。

主な機能

• コネクター（例：データベース用の JDBC）を使用して、CSV、JSON ファイル、SQL データベース、API などのソースからデータを抽出します。
• フィルタリング、集計、SQL クエリ（例：Pandas や Spark DataFrames を使用）などの処理でデータを変換します。
• スキーマ検証とエラーログ記録を行いながら、PostgreSQL、BigQuery、S3 バケットなどのターゲットにデータをロードします。
• cron のような式によるパイプラインのスケジュール設定や、Airflow などのオーケストレーションツールとの統合をサポートします。
• 最後に処理したタイムスタンプや変更データキャプチャ（CDC）を追跡して、増分ロードを処理します。

使用パターン

このスキルを使用するには、特定の ETL コマンドを指定して OpenClaw を呼び出します。まず JSON 形式でパイプラインの設定を定義し、次に CLI または API 経由で実行します。たとえば、次のような設定ファイルを渡します。

{
  "source": {"type": "file", "path": "data/input.csv"},
  "transform": {"operations": ["filter column='id' > 100"]},
  "destination": {"type": "postgres", "table": "processed_data"}
}

パイプラインはモジュール式に構成します。最初に抽出し、次にメモリ内または分散環境で変換し、最後に再試行の仕組みを使ってロードします。認証用の環境変数は必ず設定します（例：export OPENCLAW_API_KEY=your_key）。

主なコマンド/API

ETL タスクには OpenClaw CLI を使用します。たとえば、次のように実行します。

• パイプラインを作成：openclaw etl create --config path/to/config.json --env $OPENCLAW_API_KEY （フラグ：--config は JSON ファイル、--env は認証に使用します。パイプライン ID が出力されます。）
• パイプラインを実行：openclaw etl run <pipeline-id> --params '{"batch_size": 1000}' （フラグ：--params は実行時の上書きに使用します。stdout で進行状況を監視します。）
• API エンドポイント：POST /v1/etl/pipelines で作成します（リクエスト本文は JSON 設定）。GET /v1/etl/pipelines/{id} でステータスを取得します。 コードに統合するには、OpenClaw の Python SDK を使用します。

import openclaw
client = openclaw.Client(api_key=os.environ['OPENCLAW_API_KEY'])
pipeline = client.etl.create(config={'source': 'file.csv', 'transform': 'sql_query'})
client.etl.run(pipeline.id)

実行前に、必ず openclaw etl validate --file path/to/config.json で設定を検証します。

統合に関する注意

設定で依存関係を指定するなどして、このスキルをデータツールと統合します。たとえば、分散処理には "engine": "spark" を指定します。AWS では、$AWS_ACCESS_KEY_ID などの環境変数を設定し、ソースには S3 などのコネクターを使用します。出力を渡して他の OpenClaw スキルと連携させることもできます（例：ETL の結果を機械学習スキルにパイプで渡す）。データ形式（例：ビッグデータには Parquet）を合わせ、互換性を確保します。複数のツールを組み合わせる場合は、Webhook を使用します。外部イベントをきっかけに実行するには、POST /v1/etl/webhook を設定します。

エラー処理

コマンドを try-catch ブロックで囲むか、一時的な障害（例：ネットワークの問題）発生時に自動で再試行する --retry 3 などの組み込みフラグを使用して、エラーに対処します。API のレスポンスでエラーコード（例：不正な設定を示す 400、サーバーエラーを示す 500）を確認し、詳細をログに記録します。コードでは、次のようにします。

try:
    client.etl.run(pipeline.id)
except openclaw.EtlError as e:
    print(f"Error: {e.code} - {e.message}; Retrying...")

openclaw etl logs <pipeline-id> でログを監視し、失敗に対するしきい値を設定します（例：レコードの 10% を超えて検証に失敗した場合は中止）。$ETL_ERROR_WEBHOOK_URL などの環境変数を使用して、カスタムのエラーハンドラーを設定します。

具体的な使用例

1. CSV ファイルから抽出し、SQL で変換して PostgreSQL に読み込む：
   まず、設定を作成します：{"source": {"type": "file", "path": "sales.csv"}, "transform": {"sql": "SELECT * FROM data WHERE amount > 100"}, "destination": {"type": "postgres", "table": "sales_filtered", "conn_str": "dbname=mydb"}}
   次に、実行します：openclaw etl create --config sales_config.json; openclaw etl run <id> --env $OPENCLAW_API_KEY
   標準的な環境では、100 万行を 5 分以内に処理できます。

2. データベースをソースとする BigQuery 向けの増分 ETL：
   設定：{"source": {"type": "mysql", "query": "SELECT * FROM orders WHERE updated_at > '2023-01-01'"}, "transform": {"operations": ["add_column": "processed_at=NOW()"]}, "destination": {"type": "bigquery", "dataset": "my_dataset", "table": "orders_incremental"}}
   実行：openclaw etl run <pipeline-id> --params '{"incremental": true}'
   日次更新に対応し、新しいレコードのみを追加します。

グラフ上の関係

• 関連：データエンジニアリングのクラスター（例：データウェアハウジングやビッグデータ処理などのスキル）
• タグによる関連付け：etl（data-pipelines にリンク）、data-engineering（analytics-tools にリンク）
• 依存関係：$OPENCLAW_API_KEY による認証が必要です。オーケストレーションには Spark や Airflow などの外部ツールと連携します。