BigQuery への変換のプッシュダウン

このページでは、Cloud Data Fusion で Spark の代わりに BigQuery に JOIN 変換を push する方法について説明します。

詳細については、変換のプッシュダウンの概要をご覧ください。

始める前に

変換のプッシュダウンは、バージョン 6.5.0 以降で使用できます。パイプラインが以前の環境で実行されている場合は、インスタンスを最新バージョンにアップグレードできます。

パイプラインで変換のプッシュダウンを有効にする

Console

  1. インスタンスページに移動します。

    [インスタンス] に移動

  2. 目的のインスタンスを開き、Cloud Data Fusion UI の [Pipeline List] ページに移動します。

  3. 目的のデプロイ済みパイプラインを選択して、Pipeline Studio で開きます。

  4. [構成] をクリックします。

  5. [変換のプッシュダウン] タブを選択します。

    変革のプッシュダウンを有効にする

  6. [変換のプッシュダウンを有効にする] をクリックします。

  7. 変換のプッシュダウンを構成します。

    • BigQuery データセット(必須): BigQuery 実行に使用される一時テーブルのステージングに使用される BigQuery データセットの名前。データセットは、カスタマー プロジェクト内に含まれています。データセットは、テーブルとビューへのアクセスを整理して制御するために使用される最上位のコンテナです。

    • サービス アカウント タイプ(省略可): 次のいずれかを選択します。

      • ファイルパス: サービス アカウントが配置されているファイルパス。
      • JSON: サービス アカウントの JSON コンテンツ。
    • サービス アカウント ファイルパス(省略可): 承認に使用するサービス アカウント キーのローカル ファイル システム上のパス。Dataproc クラスタで実行する場合は、auto-detect に設定できます。他のクラスタで実行する場合、クラスタ内の各ノードにそのファイルが存在する必要があります。デフォルトは auto-detect です。

    • サービス アカウント JSON(省略可): サービス アカウント JSON ファイルの内容。

    • プロジェクト ID(省略可): プロジェクトを一意に識別する Google Cloud プロジェクト ID。Console のダッシュボードにあります。デフォルトでは、インスタンスが実行されている現在のプロジェクトになります。

    • 一時的なバケット名(省略可): 一時データを保存する Google Cloud Storage バケット。存在しない場合は自動的に作成されますが、自動的には削除されません。BigQuery に読み込まれた Cloud Storage データは削除されます。指定されていない場合は、一意のバケットが作成され、パイプラインの実行が完了した後に削除されます。サービス アカウントには、構成されたプロジェクトでバケットを作成する権限が必要です。

    • ロケーション(省略可): BigQuery データセットを作成するロケーション名を入力します。データセットまたは一時バケットがすでに存在する場合、この値は無視されます。デフォルト値は US multi-region です。

    • 完了後に BigQuery テーブルを保持する(省略可): パイプラインの実行中に作成されたすべての BigQuery 一時テーブルを保持する場合に選択します。これは、デバッグや検証の目的で使用できます。デフォルトは No です。

    • 一時テーブル TTL(時間)(省略可): BigQuery 一時テーブルのテーブル TTL を時間数で設定します。パイプラインが急にキャンセルされ、クリーンアップ プロセスが中断された場合(実行クラスタが突然シャットダウンされた場合など)にフェイルセーフとして使用されます。この値を 0 に設定すると、テーブル TTL が無効になります。デフォルト値は 72 (3 日)です。

    • ジョブの優先度(省略可): BigQuery ジョブの実行に使用される優先度。次のオプションのいずれかを選択します。

      • バッチ: バッチジョブはキューに入れられ、アイドル状態のリソースが使用可能になると直ちに(通常は数分以内に)開始されます。ジョブが 3 時間以内に開始されない場合、優先度は interactive に切り替えられます。

      • インタラクティブ: インタラクティブ ジョブは可能な限り早く実行され、同時実行レート上限と 1 日あたりのレート制限にカウントされます。

ログでパフォーマンスの変化をモニタリングする

パイプラインのランタイムログには、BigQuery で実行される SQL クエリを示すメッセージが含まれます。パイプラインのどのステージが BigQuery に push されるかをモニタリングできます。

次のログの例は、パイプラインの実行開始時のログエントリを示しています。このログは、パイプラインでの結合が、実行のために BigQuery にプッシュされていることを示しています。

  INFO  [Driver:i.c.p.g.b.s.BigQuerySQLEngine@190] - Validating join for stage 'Users' can be executed on BigQuery: true
  DEBUG [batch-sql-engine-adapter:i.c.c.e.s.b.BatchSQLEngineAdapter@131] - Starting push for dataset 'UserProfile'
  DEBUG [batch-sql-engine-adapter:i.c.c.e.s.b.BatchSQLEngineAdapter@131] - Starting push for dataset 'UserDetails'
  DEBUG [batch-sql-engine-adapter:i.c.c.e.s.b.BatchSQLEngineAdapter@292] - Starting join for dataset 'Users'
  INFO  [Driver:i.c.p.g.b.s.BigQuerySQLEngine@190] - Validating join for stage 'UserPurchases' can be executed on BigQuery: true
  DEBUG [batch-sql-engine-adapter:i.c.c.e.s.b.BatchSQLEngineAdapter@131] - Starting push for dataset 'Purchases'
  DEBUG [batch-sql-engine-adapter:i.c.c.e.s.b.BatchSQLEngineAdapter@292] - Starting join for dataset 'UserPurchases'
  INFO  [Driver:i.c.p.g.b.s.BigQuerySQLEngine@190] - Validating join for stage 'MostPopularNames' can be executed on BigQuery: true
  DEBUG [batch-sql-engine-adapter:i.c.c.e.s.b.BatchSQLEngineAdapter@131] - Starting push for dataset 'FirstNameCounts'
  DEBUG [batch-sql-engine-adapter:i.c.c.e.s.b.BatchSQLEngineAdapter@292] - Starting join for dataset 'MostPopularNames'
  DEBUG [batch-sql-engine-adapter:i.c.c.e.s.b.BatchSQLEngineAdapter@193] - Starting pull for dataset 'MostPopularNames'

次の例は、プッシュダウン実行に含まれる各データセットに割り当てられるテーブル名を示しています。

  INFO  [batch-sql-engine-adapter:i.c.p.g.b.s.BigQuerySQLEngine@145] - Executing Push operation for dataset Purchases stored in table <TABLE_ID>
  INFO  [batch-sql-engine-adapter:i.c.p.g.b.s.BigQuerySQLEngine@145] - Executing Push operation for dataset UserDetails stored in table <TABLE_ID>
  INFO  [batch-sql-engine-adapter:i.c.p.g.b.s.BigQuerySQLEngine@145] - Executing Push operation for dataset FirstNameCounts stored in table <TABLE_ID>
  INFO  [batch-sql-engine-adapter:i.c.p.g.b.s.BigQuerySQLEngine@145] - Executing Push operation for dataset UserProfile stored in table <TABLE_ID>

実行が進むと、ログにプッシュ ステージの完了と、最終的に結合オペレーションの実行が表示されます。次に例を示します。

  DEBUG [batch-sql-engine-adapter:i.c.c.e.s.b.BatchSQLEngineAdapter@133] - Completed push for dataset 'UserProfile'
  DEBUG [batch-sql-engine-adapter:i.c.c.e.s.b.BatchSQLEngineAdapter@133] - Completed push for dataset 'UserDetails'
  DEBUG [batch-sql-engine-adapter:i.c.p.g.b.s.BigQuerySQLEngine@235] - Executing join operation for dataset Users
  INFO  [batch-sql-engine-adapter:i.c.p.g.b.s.BigQueryJoinDataset@118] - Creating table `<TABLE_ID>` using job: <JOB_ID> with SQL statement: SELECT `UserDetails`.id AS `id` , `UserDetails`.first_name AS `first_name` , `UserDetails`.last_name AS `last_name` , `UserDetails`.email AS `email` , `UserProfile`.phone AS `phone` , `UserProfile`.profession AS `profession` , `UserProfile`.age AS `age` , `UserProfile`.address AS `address` , `UserProfile`.score AS `score` FROM `your_project.your_dataset.<DATASET_ID>` AS `UserProfile` LEFT JOIN `your_project.your_dataset.<DATASET_ID>` AS `UserDetails` ON `UserProfile`.id = `UserDetails`.id
  INFO  [batch-sql-engine-adapter:i.c.p.g.b.s.BigQueryJoinDataset@151] - Created BigQuery table `<TABLE_ID>
  INFO  [batch-sql-engine-adapter:i.c.p.g.b.s.BigQuerySQLEngine@245] - Executed join operation for dataset Users

すべてのステージが完了すると、Pull オペレーションが完了したことを示すメッセージが表示されます。これは、BigQuery のエクスポート プロセスがトリガーされ、このエクスポート ジョブの開始後、レコードのパイプラインへの読み取りが開始されることを示します。次に例を示します: 「

DEBUG [batch-sql-engine-adapter:i.c.c.e.s.b.BatchSQLEngineAdapter@196] - Completed pull for dataset 'MostPopularNames'

パイプラインの実行でエラーが発生した場合は、ログに表示されます。

BigQuery JOIN オペレーションの実行に関する追加情報(リソース使用率、実行時間、エラーの原因など)を確認するには、表示されているジョブ ID を使用して BigQuery ジョブデータを表示します。これはジョブログにあります。

次のステップ