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

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

このプロセスでは、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 カタログ、名前空間、テーブルを作成します。

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

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

サポートと制限事項

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

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

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

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

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

    [レイクハウス] に移動

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

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

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

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

    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. [完了] をクリックします。

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

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

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

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

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

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

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

    4. Amazon S3 または Azure Blob Storage から移行する場合は、Parquet ファイルを Cloud Storage にコピーするための追加の構成を指定する必要があります。ファイルがすでに Cloud Storage にある場合は、この操作は必要ありません。

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

      2. gcs_file_path 構成パラメータの値を設定して、一時ファイルのコピー先となる完全修飾 Cloud Storage バケットを指定します。Lakehouse ウェアハウスで使用されている Cloud Storage バケットと同じものを使用することをおすすめします。

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

      1 [Dataflow オプション] セクションをクリックして開きます。

      1. Parquet ファイルが Amazon S3 にある場合は、[追加のパイプライン オプションを追加] をクリックして、Apache Beam の S3 関連のパイプライン オプションを指定します。たとえば、s3_region_name、s3_access_key_id、s3_secret_access_key と、それに対応する値などです。

      2. Parquet ファイルが Azure Blob Storage にある場合は、[追加のパイプライン オプションを追加] をクリックして、Apache Beam の Azure 関連のパイプライン オプションを指定します。たとえば、azure_connection_string、blob_service_endpoint、azure_managed_identity_client_id と、それに対応する値などです。

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

    1. [書き込み結果] シンクをクリックして開きます。

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

      gs://BUCKET_NAME/errors/errors.json

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

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

次のステップ