Dataflow ジョブビルダーを使用して Delta Lake テーブルを Borderless Lakehouse にインポートする

大量のデータを移動する必要がある場合、Delta Lake テーブルをボーダレスな Lakehouse に移行するのは複雑で費用がかかる可能性があります。Dataflow ジョブビルダーは、Cloud Storage Delta Lake テーブルを Lakehouse に直接インポートするためのローコードまたはノーコードのインターフェースを提供します。既存のデータは、データの量に基づいて自動スケーリングする Dataflow ジョブを使用して、Delta Lake テーブルからターゲット Lakehouse テーブルにコピーされます。

新しいテーブルの場合、Dataflow はスキーマを自動的に作成します。既存のテーブルの場合、スキーマは変更されないため、ジョブを成功させるには、宛先テーブルのスキーマがコピー元 Delta Lake テーブルに正しくマッピングされている必要があります。

次の接続の詳細を使用して、Cloud Storage に保存されている Delta Lake テーブルからデータをインポートします。

このデータをインポートした後、継続的移行を有効にすることができます。

始める前に

Delta Lake テーブルのデータをインポートするには、次のものが必要です。

  1. Dataflow API、BigQuery API、Lakehouse API が有効になっていない場合は、有効にします。

    API を有効にするために必要なロール

    API を有効にするには、serviceusage.services.enable 権限が必要です。プロジェクトを作成した場合は、オーナーロール（roles/owner）を介してこの権限がすでに付与されている可能性があります。それ以外の場合は、Service Usage 管理者ロール（roles/serviceusage.serviceUsageAdmin）を介してこの権限を取得できます。ロールを付与する方法を確認する。

    API を有効にする

  2. リソースの作成に必要な権限を取得するには、プロジェクトに必要な Identity and Access Management（IAM）ロールを付与するよう管理者に依頼してください。

  3. Cloud Storage バケットに保存されている既存の Delta Lake テーブル。テーブル ディレクトリは、Parquet データファイルと _delta_log/ トランザクション ログ ディレクトリを含む有効な Delta Lake テーブルのルートである必要があります。

  4. データをインポートする Lakehouse Iceberg カタログ、名前空間、テーブル。

サポートと制限事項

Dataflow を使用して Delta Lake テーブルデータを Lakehouse にインポートする場合、次の制限があります。

  • この機能を使用するには、バッチ パイプライン ジョブを使用する必要があります。
  • 既存の宛先テーブルの場合、スキーマは変更されません。宛先テーブルのスキーマは、ソースの Delta Lake テーブルのスキーマに正しくマッピングされている必要があります。
  • ソースデータは、Cloud Storage に保存されている有効な Delta Lake テーブルである必要があります。テーブルのルート ディレクトリには、Delta Lake によって作成された Parquet データファイルと _delta_log/ トランザクション ログ ディレクトリ（JSON または Parquet ログファイルを含む）が含まれている必要があります。
  • Amazon S3 は Delta Lake テーブルソースではサポートされていません。

Delta Lake テーブルをインポートする

Delta Lake テーブルを Lakehouse にインポートするには、次の操作を行います。

  1. Google Cloud コンソールで、[Lakehouse ランタイム カタログ] ページに移動します。

    Lakehouse ランタイム カタログに移動

  2. データをインポートするカタログ、名前空間、テーブルを選択します。

  3. [テーブルの詳細] ページで、 [テーブルをインポート] をクリックし、[Delta Lake から（バッチ）] を選択します。

    Dataflow の [ジョブビルダー] ページが開き、Delta Lake から Lakehouse へのブループリントが読み込まれます。

  4. [ソース] セクションで、次の操作を行います。

    1. [ReadFromDeltaLake Delta Lake テーブル] ソースパネルを開くには、展開矢印 をクリックします。

    2. [テーブルパス] フィールドに、Delta Lake テーブルのルート ディレクトリ（データファイルと _delta_log/ ディレクトリを含むディレクトリ）の Cloud Storage URI を入力します。例: gs://BUCKET_NAME/tables/TABLE_NAME

    3. 省略可: [Hadoop 構成プロパティ] フィールドで、Cloud Storage からの読み取りに必要な追加の Hadoop 構成プロパティを構成します。例: fs.gs.impl=com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem。

    4. [完了] をクリックします。

  5. [シンク] セクションで、次の操作を行います。

    1. 省略可: WriteToIceberg Lakehouse テーブル シンクパネルを確認します。通常、このパネルの情報（Lakehouse テーブル、カタログ名、ウェアハウスの場所など）は事前入力されています。

    2. [完了] をクリックします。

  6. [Dataflow オプション] セクションで、[ジョブを実行] をクリックします。

Delta Lake テーブルのインポートに使用される Dataflow パイプラインをさらにカスタマイズする必要がある場合は、ジョブビルダー フォームまたは YAML エディタを使用してカスタマイズできます。

ジョブの出力を調べる

ジョブが完了したら、BigQuery でクエリを実行して、データが Lakehouse テーブルにコピーされたことを確認できます。

  1. Dataflow ジョブのリストで、ジョブのステータスが [成功] であることを確認します。

    [ジョブ] に移動

  2. ジョブが失敗した場合やエラーが発生した場合は、ジョブログまたはワーカーログで詳細を確認します。

  3. Google Cloud コンソールで、BigQuery Studio ページに移動します。

    BigQuery に移動

  4. クエリエディタで、テーブルを検査する SQL クエリを入力します。PROJECT_ID.CATALOG.NAMESPACE.TABLE_NAME 規約を使用して、次のクエリを実行できます。

    SELECT * FROM `PROJECT_ID`.`CATALOG`.`NAMESPACE`.`TABLE_NAME` LIMIT 10;

  5. （実行）をクリックします。

  6. クエリ結果を確認して、データが正しく処理されたことを確認します。

省略可。継続的移行を有効にする

Delta Lake から Lakehouse へのエンドツーエンドの継続的移行を設定するには、次の操作を順に行います。

  1. 入力 Delta Lake テーブルで delta.enableChangeDataFeed プロパティを有効にします。
  2. delta.enableChangeDataFeed プロパティを設定した後、commit ID またはタイムスタンプにテーブルの初期移行を行います。
  3. ワークロードの要件に合わせて選択した頻度（コミット ID またはタイムスタンプ範囲）を使用して、後続のデータの CDC ジョブをスケジュール設定して実行します。

詳細については、Dataflow を使用して Delta Lake の変更データ キャプチャ（CDC）データを Lakehouse に移行するをご覧ください。

次のステップ