Dataflow を使用してストレージから Lakehouse に Parquet ファイルをインポートする

Dataflow ジョブビルダーのブループリントを使用すると、クラウドベースのストレージ（Cloud Storage または Amazon S3）から Lakehouse の Apache Iceberg テーブルに既存の Apache Parquet ファイルを追加できます。

このプロセスでは、IcebergAddFiles 変換を使用します。Parquet ファイルが Cloud Storage にある場合、この変換では、基盤となるデータを移動または書き換えることなく、Lakehouse にファイルを登録します。ファイルが Amazon S3 などの外部ストレージ システムにある場合は、Lakehouse を介して高速にクエリを実行できるように Cloud Storage にコピーされてから登録されます。

次の接続の詳細を使用して、クラウドベースのストレージから Lakehouse の Apache Iceberg テーブルに Parquet ファイルを追加します。

始める前に

  1. Dataflow API、BigQuery API、Lakehouse API を有効にします。

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

  3. データをインポートする Lakehouse for Apache Iceberg カタログ、Namespace、テーブルを作成します。

  4. クラウドベースのストレージ バケット（Cloud Storage または Amazon S3）を作成し、Parquet ファイルをバケットにアップロードします。

  5. 使用しているクラウドベースのストレージ バケットが Google の Cloud Storage ではない場合は、ジョブのエラーログを保存する Cloud Storage バケットを作成します。

サポートと制限事項

Dataflow を使用してクラウドベースのストレージにある Parquet ファイルを Lakehouse for Apache Iceberg にインポートする場合、次の制限があります。

  • ソースデータは Apache Parquet 形式で、Cloud Storage または Amazon S3 に保存されている必要があります。
  • この機能はバッチ パイプラインのみをサポートしています。

Parquet ファイルを Lakehouse にインポートする

次の手順に沿って、Dataflow ジョブビルダー UI を使用して、クラウドベースのストレージから Lakehouse の Iceberg テーブルに Parquet ファイルをインポートします。

  1. Google Cloud コンソールで、[Lakehouse for Apache Iceberg] ページに移動します。

    [Lakehouse] に移動

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

  3. [テーブルの詳細] ページで、 [テーブルをインポート] をクリックします。

  4. [インポート構成] ダイアログで、[Apache Parquet ファイルから Lakehouse にテーブルをインポート（バッチ）] を選択します。

    Dataflow の [ジョブビルダー] ページが開きます。

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

    1. すでに作成されている CreateGlobalInput ソース エントリを開きます。

    2. [YAML ソース構成] エディタ セクションで、elements シーケンスに Parquet ファイルの 1 つ以上のパスを入力します。

      インポートの効率を高めるには、多数のファイルを登録する際に、複数のファイルセット（glob）を指定します。次に例を示します。

      reshuffle: true
elements:
  -   gs://BUCKET_NAME/restaurant-data/2023/*.parquet
  -   gs://BUCKET_NAME/restaurant-data/2024/*.parquet

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

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

    1. [IcebergAddFiles] 変換セクションをクリックして開きます。

    2. [Iceberg table] フィールドに、Namespace とテーブル名を入力します。例: NAMESPACE .TABLE_NAME

    3. [カタログ プロパティ] で、次の項目を構成します。

      1. warehouse: カタログの Cloud Storage のロケーション。例: gs://CATALOG_PATH

      2. header.x-goog-user-project: Google Cloud プロジェクト ID: PROJECT_ID。

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

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

    1. [Write results] シンクをクリックして開きます。

    2. [JSON の場所] フィールドで、エラー結果を書き込む Cloud Storage の場所とファイル名を指定します。次に例を示します。

      gs://BUCKET_NAME/errors/errors.json

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

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

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

ジョブの出力を調べる

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

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

    [ジョブ] に移動

  2. ジョブが失敗した場合やエラーが発生した場合は、Cloud Storage の JSON エラーログ ファイルで詳細を確認します。

    [バケット] に移動

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

次のステップ