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

大量のデータを書き直したり移動したりする必要がある場合、Delta Lake テーブルを Lakehouse for Apache Iceberg に移行するのは複雑で費用がかかる可能性があります。基盤となるファイルを移動または書き換えることなく、Delta Lake データを Lakehouse で使用できるようにするには、Dataflow ジョブビルダーを使用します。ジョブビルダーは、Cloud Storage 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 for Apache Iceberg にインポートする場合、次の制限があります。

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

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

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

  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 でクエリを実行して、データが Iceberg テーブルに登録されたことを確認できます。

  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. クエリ結果を確認して、データが正しく処理されたことを確認します。

次のステップ