data-lakehouse

data-lakehouse

目的

このスキルを使うと、データレイクとデータウェアハウスの機能を組み合わせたデータレイクハウスアーキテクチャを設計・実装し、スケーラブルなビッグデータのストレージと分析を実現できます。Delta Lake や Iceberg などのプラットフォームで、ACID トランザクション、スキーマの進化、最適化されたクエリ性能を備えたペタバイト規模のデータを管理する際に使用します。

使用する場面

• S3 や Kafka などのソースから大規模なデータを取り込み、未加工データの保存と構造化クエリの両方が必要な場合。
• EC サイトや IoT のデータ処理における ETL パイプラインなど、リアルタイム更新が必要な分析ワークロードの場合。
• 非構造化データに対する SQL 分析のために、Spark や Presto と連携する場合。

主な機能

• アーキテクチャ設計：Iceberg などのテーブル形式を使ったパーティション戦略やメタデータ管理を含む、レイクハウス構成の設計図を生成します。
• データ取り込み：Apache Spark などのツールを使ったバッチおよびストリーミングでの取り込みをサポートし、Parquet や ORC などの形式を扱います。
• クエリ最適化：Z-order クラスタリングを使用した Delta Lake テーブルの作成など、クエリを高速化するためにキャッシュやインデックスを実装します。
• スケーラビリティ：AWS Glue による ETL ジョブなど、クラウド API を介してストレージとコンピューティングリソースを自動スケーリングします。
• セキュリティ：AWS Lake Formation の許可などのポリシーを使用して、行レベルのアクセス制御を適用します。

使用パターン

• パターン 1：新しいレイクハウスをセットアップする場合は、このスキルを呼び出して設定ファイルを生成し、それを使ってストレージを初期化します。例：CSV データから Delta Lake テーブルを作成します。
• パターン 2：分析ワークフローでは、インデックスを追加してクエリを最適化するためにこのスキルを使用し、その後 Spark SQL で実行します。
• パターン 3：メンテナンスでは、既存のテーブルに対してスキーマの進化とマージ操作のチェックを定期的に実行します。

一般的なコマンド／API

このスキルには OpenClaw CLI または API を使用します。認証するには、環境変数として $DATA_LAKEHOUSE_API_KEY を設定する必要があります。

• CLI コマンド：レイクハウスプロジェクトを初期化します：

  openclaw skill data-lakehouse init --project my-lakehouse --storage s3://my-bucket --engine delta
  

  これにより、S3 バケットと Delta Lake エンジンを含む基本的な設定ファイルが作成されます。

• API エンドポイント：POST リクエストでテーブルを作成します：

  curl -H "Authorization: Bearer $DATA_LAKEHOUSE_API_KEY" \
       -d '{"table_name": "sales_data", "format": "parquet", "partition_by": ["date"]}' \
       https://api.openclaw.ai/data-lakehouse/tables
  

  レスポンスには、すぐに使用できるテーブルのメタデータが含まれます。

• コードスニペット：Python で Spark と連携します：

  from pyspark.sql import SparkSession
  spark = SparkSession.builder.appName("lakehouse").getOrCreate()
  df = spark.read.format("delta").load("s3://my-bucket/sales_data")
  df.write.format("delta").mode("append").save("s3://my-bucket/sales_data")
  

  これにより Delta テーブルにデータが追加されます。Spark が AWS 認証情報で設定されていることを確認してください。

• 設定形式：レイクハウスの設定には JSON を使用します。例：

  {
    "storage": "s3://my-bucket",
    "engine": "iceberg",
    "auth": {"key": "$DATA_LAKEHOUSE_API_KEY"}
  }
  

  CLI で次のように読み込みます：openclaw skill data-lakehouse apply --config path/to/config.json。

統合に関する注意

• 他のスキルとの連携：「big-data」スキルに出力を渡して連携します。例：Kafka 取り込みスキルからのデータを Spark ストリーミングでこのスキルにパイプします。
• 外部ツール：バケットポリシーを設定して AWS S3 と連携します。CLI コマンドでは --aws-region us-west-2 を使用してください。Spark の場合は、spark.delta などの依存関係が環境に含まれていることを確認してください。
• API 連携：他のサービスから呼び出す場合は、レート制限に対するリトライを処理してください。例：連鎖する API 呼び出しで同じ $DATA_LAKEHOUSE_API_KEY を使用します。
• 依存関係の管理：Delta Lake との互換性のため Spark 3.0+ が必要であることを明記するなど、必ずバージョンを指定してください。

エラー処理

• 一般的なエラー：$DATA_LAKEHOUSE_API_KEY が設定されているか確認して認証エラーに対処します。スクリプトでは os.environ.get('DATA_LAKEHOUSE_API_KEY') を使用してください。
• 具体的な手順：ストレージへのアクセスエラー（例：S3 の権限エラー）が発生した場合は、try-except ブロックを追加します：

  try:
      spark.read.format("delta").load("s3://my-bucket/data")
  except Exception as e:
      print(f"Error: {e}. Check bucket permissions and retry.")
  

  API 呼び出しでは、ネットワークの問題などの一時的なエラーに対し、指数バックオフを使って再試行してください。
• デバッグ：詳細なログを表示するには CLI フラグ --verbose を使用します。例：openclaw skill data-lakehouse init --verbose。openclaw skill data-lakehouse validate --file config.json で設定を検証してください。

具体的な使用例

• 例 1：営業分析用レイクハウスの構築： まず、初期化します：openclaw skill data-lakehouse init --project sales-ana --storage s3://sales-data。次に、データを取り込みます：API を使用してテーブルを作成し、上記のように Spark でデータを追加します。最後に、次のコマンドでクエリを実行します：spark.sql("SELECT * FROM sales_data WHERE date > '2023-01-01'")。

• 例 2：IoT データ向けに既存のレイクハウスを最適化： 次を実行します：openclaw skill data-lakehouse optimize --table iot_metrics --add-index date。これによりインデックスが追加されます。クエリの例を使って確認します：df = spark.read.format("iceberg").load("s3://iot-bucket/metrics").filter(df.date > current_date())。最適化後のパフォーマンスを監視してください。

グラフ上の関係

• 関連先：「big-data」（データ処理パイプラインのタグを共有）、「data-engineering」（ETL ワークフローでクラスターに所属）。
• 接続方法：「data-lakehouse」などのタグを使ったスキル横断クエリや、順次実行するタスクフロー向けの「data-engineering」クラスター。