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

Dataflow ジョブビルダーを使用すると、ローコードまたはノーコードのインターフェースを介して Cloud Storage Delta Lake テーブルをボーダーレス Lakehouse に直接インポートできるため、大規模なデータ移行の複雑さが軽減されます。自動スケーリング Dataflow バッチジョブは、データを Lakehouse のターゲット Apache Iceberg テーブルにコピーし、新しいテーブルのスキーマを自動的に作成します。既存のテーブルには、一致するスキーマが必要です。

データをインポートしたら、必要に応じて継続的な移行を有効にできます。

始める前に

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 to 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. ワークロードの要件に合わせて選択した頻度（commit ID またはタイムスタンプ範囲）を使用して、後続のデータに対してスケジュールされた CDC ジョブを実行します。

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

次のステップ