Data Engineering Agent を使用したデータ パイプラインの構築と変更

プレビュー版

この機能は、 サービス固有の規約の対象となります。 pre-GA の機能は「現状のまま」で提供されるものであり、サポートが制限されることがあります。 詳細については、リリース ステージの説明をご覧ください。

注: このプレビュー機能に関するフィードバックやご質問については、[e-mail] までお問い合わせください。

Data Engineering Agent を使用すると、Gemini in BigQuery を使用して、BigQuery でデータを読み込んで処理するデータ パイプラインを構築、変更、管理できます。Data Engineering Agent を使用すると、自然言語プロンプトを使用して、さまざまなデータソースからデータ パイプラインを生成したり、データ エンジニアリングのニーズに合わせて既存のデータ パイプラインを調整できます。Data Engineering Agent には次の機能があります。

• パイプライン作成用の自然言語: エージェントは Gemini を使用してデータを理解し、平易な言語の指示を解釈します。平易な言葉で指示することで、Data Engineering Agent にデータ パイプラインの構築や編集を行わせることができます。

• Dataform の統合: エージェントは、必要なパイプライン コードを生成して、Dataform リポジトリ内の SQLX ファイルに整理します。エージェントが Dataform ワークスペースで動作するため、Dataform パイプラインはエージェントで自動的に使用できます。

• カスタム エージェントの手順: 平易な言語でエージェントの手順を作成して、Data Engineering Agent のカスタムルールを定義します。エージェントの指示は組織全体で永続的に適用され、命名規則やスタイルガイドなど、組織全体のルールを適用する際に役立ちます。

• パイプラインの検証: エージェントは、生成されたコードを検証して、データ パイプラインが機能していることを確認します。

Data Engineering Agent で自然言語プロンプトを使用して、テーブル、ビュー、アサーション、宣言、オペレーションの SQLX ファイルを作成できます。たとえば、Data Engineering Agent を使用して次のことができます。

• Cloud Storage などの外部データソースから、CSV、AVRO、PARQUET などのさまざまな形式でデータを読み込みます。
• 既存の BigQuery ルーティン（UDF）を作成または使用して、データに対してカスタム分析と変換を行います。
• エージェントに再利用可能なガイドラインを自然言語で定義します。

Data Engineering Agent で使用できるプロンプトの例については、プロンプトの例をご覧ください。

制限事項

Data Engineering Agent には次の制限があります。

• Data Engineering Agent は pre-GA サービスであり、本番環境での使用は想定されていません。
• Data Engineering Agent は、次のファイル形式の自然言語コマンドをサポートしていません。
  • ノートブック
  • データの準備
  • SQLx 内の JavaScript
• Data Engineering Agent はパイプラインを実行できません。ユーザーは、パイプラインを確認して実行またはスケジュールする必要があります。
• Data Engineering Agent は、存在しない中間リソースに依存する SQL を、パイプラインの完全な呼び出し（ユーザー トリガー）なしで検証できません。
• Data Engineering Agent は、手順や直接プロンプトで提供されたウェブリンクや URL を検索できません。
• エージェント指示ファイルでファイルをインポートする場合、@ インポート構文は、./、/、または文字で始まるパスのみをサポートします。
• データ プレビュー機能は、hasOutput フラグが true に設定されているテーブル、宣言、クエリでのみサポートされます。

サポートされるリージョン

Gemini in BigQuery はロード バランシングに Google Cloudを使用するため、グローバルに運用されます。Gemini in BigQuery はグローバルに運用されるため、使用するリージョンを選択できません。

Dataform と BigQuery のすべてのリクエストは、それぞれのリージョン エンドポイントに送信されるため、データとコードはリージョン内に留まります。

Gemini in BigQuery は、次のリージョンから提供されます。

南北アメリカ

• アイオワ（us-central1）

ヨーロッパ

• フィンランド（europe-north1）
• フランクフルト（europe-west3）

Data Engineering Agent によるデータの使用方法

Data Engineering Agent は、より高品質なエージェント レスポンスを生成するために、BigQuery テーブルのサンプル行や Dataplex Universal Catalog で生成されたデータ スキャン プロファイルなど、BigQuery と Dataplex Universal Catalog から追加のデータとメタデータを取得できます。このデータは Data Engineering Agent のトレーニングには使用されず、エージェントの会話中にのみ、エージェントの回答を通知するための追加のコンテキストとして使用されます。

始める前に

Google Cloud プロジェクトで Gemini in BigQuery が有効になっていることを確認します。詳細については、Gemini in BigQuery を設定するをご覧ください。

Gemini Data Analytics API も有効にする必要があります。

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the API

必要なロール

Data Engineering Agent の使用に必要な権限を取得するには、プロジェクトに対する次の IAM ロールを付与するよう管理者に依頼してください。

• Dataform コードエディタ（roles/dataform.codeEditor）
• BigQuery ジョブユーザー（roles/bigquery.jobuser）

ロールの付与については、プロジェクト、フォルダ、組織へのアクセス権の管理をご覧ください。

必要な権限は、カスタムロールや他の事前定義ロールから取得することもできます。

Data Engineering Agent を使用してデータ パイプラインを生成する

BigQuery で Data Engineering Agent を使用するには、次のいずれかのオプションを選択します。

BigQuery パイプライン

BigQuery パイプライン インターフェースで Data Engineering Agent を使用するには、次の操作を行います。

1. [BigQuery] ページに移動します。

   BigQuery に移動

2. クエリエディタで、arrow_drop_down [新規作成] > [パイプライン] の順にクリックします。

3. 実行認証情報のオプションを選択し、[開始] をクリックします。これらの認証情報はエージェントでは使用されませんが、生成されたデータ パイプラインを実行するために必要です。

4. [Try out the agent experience for data pipeline] をクリックします。

5. [エージェントに質問] フィールドに、データ パイプラインを生成するための自然言語プロンプトを入力します。例:

     Create dimension tables for a taxi trips star schema from
     new_york_taxi_trips.tlc_green_trips_2022. Generate surrogate keys and all
     the descriptive attributes.
   

   プロンプトを入力したら、[送信] をクリックします。

6. Data Engineering Agent は、プロンプトに基づいてデータ パイプラインを生成します。

Data Engineering Agent によって生成されたデータ パイプラインは、データ パイプラインの提案されたドラフトです。パイプライン ノードをクリックすると、生成された SQLX クエリを確認できます。エージェントが提案したデータ パイプラインを適用するには、[適用] をクリックします。

Dataform

Dataform で Data Engineering Agent を使用するには、次の操作を行います。

1. [Dataform] ページに移動します。

   Dataform に移動

2. リポジトリを選択します。

3. 開発ワークスペースを作成または選択します。

4. ワークスペースで、[エージェントに質問] をクリックします。

5. 表示された [エージェントに質問] プロンプトに、データ パイプラインを生成するための自然言語プロンプトを入力します。例:

     Create dimension tables for a taxi trips star schema from
     new_york_taxi_trips.tlc_green_trips_2022. Generate surrogate keys and all
     the descriptive attributes.
   

   プロンプトを入力したら、[送信] をクリックします。

プロンプトが送信されると、Data Engineering Agent はプロンプトに基づいてデータ パイプラインを生成し、Dataform SQLX ファイルを変更します。エージェントは、これらの変更をワークスペース ファイルに直接適用します。

データ パイプラインを確認する

Data Engineering Agent によって生成されたデータ パイプラインで、パイプライン ノードをクリックして確認できます。

• [構成] タブには、ノードに関連付けられている生成済みの SQLX クエリが表示されます。
• [データ プレビュー] タブには、ファイルの入力テーブルと出力テーブルが表示されます。このノードでデータ変換のプレビューを表示するには、[タスクを実行] をクリックして、依存関係の有無にかかわらずタスクを実行します。

データ パイプラインを編集する

データ パイプラインを編集するには、Data Engineering Agent を使用します。[エージェントに質問] をクリックし、データ パイプラインの変更を提案するプロンプトを入力します。

Data Engineering Agent によって提案された変更を確認し、[適用] をクリックして変更を適用します。

パイプライン ノードを選択して [開く] をクリックすると、SQLX クエリを手動で編集することもできます。

Data Engineering Agent のエージェントへの指示を作成する

エージェントへの指示は、Data Engineering Agent に対する自然言語の指示です。この指示を使用すると、永続的な指示を保存して、エージェントがカスタムの事前定義された一連のルールに従うようにできます。組織全体でエージェントの結果を統一したい場合（命名規則やスタイルガイドの適用など）は、エージェントへの指示を使用します。

Data Engineering Agent のエージェント指示ファイルとして GEMINI.MD コンテキスト ファイルを作成できます。ローカル ワークスペースで使用するエージェント指示ファイルを作成することも、外部リポジトリを使用して複数のデータ パイプラインで同じ指示ファイルを使用することもできます。

次の手順でエージェントへの指示を作成します。

1. [エージェントに質問] で、[Pipeline instructions] をクリックします。
2. [パイプラインの指示ペイン] で、[Create instructions file] をクリックします。

3. 表示された GEMINI.MD ファイルに、自然言語で指示を入力します。

   次の例は、複数のルールを含むエージェント指示ファイルを示しています。

     1. All event-specific tables MUST be prefixed with `cs_event_`.
     2. The primary key for any player activity table is a composite key of `player_id` and `event_timestamp_micros`.
     3. Filter out any player actions where `mana_spent` is greater than `max_mana_pool`. This is considered a data anomaly.
   

4. [保存] をクリックします。

エージェント指示ファイルを最適に構成する方法については、エージェント指示ファイルのベスト プラクティスをご覧ください。

外部リポジトリからエージェントへの指示を読み込む

外部リポジトリをリンクすることで、複数のデータ パイプラインで再利用されるエージェンへの指示セットを指定できます。

1. [エージェントに質問] で、[Pipeline instructions] をクリックします。
2. [外部リポジトリ] で、[外部リポジトリの手順を使用する] を選択します。
3. 指定されたフィールドに、データ パイプラインで使用するエージェントの指示を含むリポジトリを指定します。
4. [保存] をクリックします。

追加のローカル ファイルをエージェントへの指示としてインポートする

@file.md 構文を使用して、Data Engineering Agent の他の指示ファイルを GEMINI.md ファイルにインポートすることもできます。詳細については、メモリ インポート プロセッサをご覧ください。

データ パイプライン エラーのトラブルシューティング

データ パイプラインの生成中にエラーが発生した場合は、Data Engineering Agent の実行に必要な前提条件をすべて満たしていることを確認します。詳細については、始める前にをご覧ください。

生成されたデータ パイプラインでエラーが発生した場合は、次の手順で Data Engineering Agent にデータパイプラインの障害を診断し、トラブルシューティングの推奨事項を提案するよう求めることができます。

1. パイプラインまたは開発ワークスペースで、[実行] タブをクリックします。

2. 実行リストから、失敗したデータ パイプラインの実行を見つけます。失敗した実行は、実行の [ステータス] 列で確認できます。

   

3. アイコンにカーソルを合わせ、[調査] をクリックします。Data Engineering Agent は、データ パイプラインの実行でエラーが発生していないかどうかを分析します。

   

4. 分析が完了すると、Data Engineering Agent は [Observations and Hypothesis] セクションにレポートを生成します。このレポートには、以下が含まれます。

   • データ パイプラインの実行ログから抽出されたモニタリング結果とデータポイント
   • 考えられる失敗の原因
   • 特定された問題を解決するための行動につながる手順または推奨事項のセット

Data Engineering Agent のトラブルシューティング レポートを使用すると、推奨事項を手動で実装できます。次の手順で、Data Engineering Agent に修正の適用を指示することもできます。

1. トラブルシューティング レポートの提案をコピーします。
2. Data Engineering Agent に戻ります。
   1. BigQuery パイプラインを使用している場合は、パイプライン ページに移動して [エージェントに質問] をクリックします。
   2. Dataform を使用している場合は、[エージェントに質問] をクリックします。
3. 提案をプロンプトに貼り付け、Data Engineering Agent にデータ パイプラインを直接修正するよう指示します。
4. [送信] をクリックします。

サンプル プロンプト

以降のセクションでは、Data Engineering Agent を使用してデータ パイプラインを開発する際に使用できるサンプル プロンプトを示します。

既存のデータを集計して新しいテーブルにする

このプロンプトを使用すると、Data Engineering Agent はスキーマとサンプルを使用して、キーによるデータグループ化を推測します。通常、エージェントはテーブルと列の説明を含む新しいテーブル構成を設定します。

  Create a daily sales report from the
  bigquery-public-data.thelook_ecommerce.order_items table into a
  reporting.daily_sales_aggregation table.

既存のテーブルにデータ品質チェックを追加する

このプロンプトにより、エージェントはスキーマとサンプルに基づいて、テーブルの妥当な品質チェックを推測します。プロンプトの一部として、意見の分かれるアサーションを追加することもできます。

  Add data quality checks for bigquery-public-data.thelook_ecommerce.users.

新しい派生列を作成し、新しいテーブルにデータ品質チェックを追加する

次のプロンプトは、テーブルと列を追加し、同時にテーブルの品質チェックを指定する方法を示しています。

  Create a new table named staging.products from
  bigquery-public-data.thelook_ecommerce.products and add a calculated column
  named gross_profit, which is the retail_price minus the cost.


  Also, add the following assertions: ID must not be null and must be unique.
  The retail_price must be greater than or equal to the cost. The department
  column can only contain 'Men' or 'Women'.

モデル定義の一部として UDF を作成する

Data Engineering Agent は、ユーザー定義関数（UDF）を作成するように DDL を設定することもできます。エージェントは実際に UDF を作成しませんが、データ パイプラインを実行して UDF を作成できます。これらの UDF は、データ パイプラインのモデル定義で使用できます。

  Create a user-defined function (UDF) named get_age_group that takes an integer
  age as input and returns a string representing the age group ('Gen Z',
  'Millennial', 'Gen X', 'Baby Boomer').


  Use this UDF on the age column from the
  bigquery-public-data.thelook_ecommerce.users table to create a new view called
  reporting.user_age_demographics that includes user_id, age, and the calculated
  age_group.

ベスト プラクティス

以降のセクションでは、Data Engineering Agent と Dataform の操作に関するベスト プラクティスについて説明します。

Data Engineering Agent のベスト プラクティス

一般的なリクエストにはエージェントへの指示を活用する。よく使用する手法がある場合や、エージェントに同じ修正を頻繁に加える場合は、エージェントへの指示を共通の手順やリクエストを保存する一元的な場所として使用します。

エージェントに詳細なコンテキストを提供する。Dataplex Universal Catalog からエージェントに追加のコンテキストを提供するには、BigQuery のテーブルと列に用語集の用語を関連付け、データ プロファイル スキャンを生成します。用語集の用語は、特別な取り扱いが必要な個人情報（PII）を含む列など、追加のコンテキストが必要な列にタグを付ける場合や、テーブル間で名前が異なる一致する列を特定する場合に使用できます。データ プロファイル スキャンにより、エージェントはテーブルの列内のデータ分布をより深く理解し、より具体的なデータ品質アサーションを作成できます。詳細については、データのプロファイリングについてをご覧ください。

わかりやすく書く。リクエストは明確に記述します。あいまいな表現は避けましょう。可能であれば、次の例に示すように、プロンプトでソースと宛先のデータソースを指定します。

  Extract data from the sales.customers table in the us_west_1 region, and load
  it into the reporting.dim_customers table in BigQuery. Match the schema of the
  destination table.

直接的で範囲が明確なリクエストを送信する。質問は一度に 1 つにして、プロンプトを簡潔にします。複数の質問を含むプロンプトの場合は、次の例に示すように、質問の各部分を個別に列挙して、明確さを高めることができます。

  1. Create a new table named staging.events_cleaned. Use raw.events as the
     source. This new table should filter out any records where the user_agent
     matches the pattern '%bot%'. All original columns should be included.

  2. Next, create a table named analytics.user_sessions. Use
     staging.events_cleaned as the source. This table should calculate the
     duration for each session by grouping by session_id and finding the
     difference between the MAX(event_timestamp) and MIN(event_timestamp).

明示的な指示を与え、キーワードを強調する。次の例に示すように、プロンプト内の重要な用語やコンセプトを強調したり、特定の要件を重要としてラベル付けすることができます。

  When creating the staging.customers table, it is *VERY IMPORTANT* that you
  transform the email column from the source table bronze.raw_customers.
  Coalesce any NULL values in the email column to an empty string ''.

操作の順序を指定する。順序付けられたタスクの場合、次の例のように、リストでプロンプトを構成できます。リスト内の項目は、焦点を絞った小さなステップに分割します。

  Create a pipeline with the following steps:
  1. Extract data from the ecomm.orders table.
  2. Join the extracted data with the marts.customers table on customer_id
  3. Load the final result into the reporting.customer_orders table.

改善して繰り返す。さまざまな言い回しやアプローチを試して、どれが最善の結果をもたらすのか確かめます。エージェントが無効な SQL を生成したり、間違えた場合は、サンプルや公開ドキュメントを使用してエージェントをガイドします。

  The previous query was incorrect because it removed the timestamp. Please
  correct the SQL. Use the TIMESTAMP_TRUNC function to truncate the
  event_timestamp to the nearest hour, instead of casting it as a DATE. For
  example: TIMESTAMP_TRUNC(event_timestamp, HOUR).

エージェント指示ファイルに関するベスト プラクティス

エージェント指示ファイルを作成して、ニーズに合わせて Data Engineering Agent をカスタマイズします。エージェントへの指示を使用する場合は、次のことをおすすめします。

• Dataform のすべてのファイルパスは、リポジトリのルートからの相対パスです。@file.md 構文には相対パスを使用して、GEMINI.md に指示を正しくインポートします。
• GEMINI.md でインポートされたファイル自体にインポートを含めることができ、ネストされた構造を作成できます。無限再帰を防ぐため、GEMINI.md の最大インポート深度は 5 レベルになっています。
• データ パイプライン間で指示を共有するには、指示を一元的な Dataform リポジトリに保存し、作業用の Dataform リポジトリにリンクします。ローカル指示を使用すると、パイプライン固有の動作に対して中央のルールをオーバーライドできます。
• エージェントの指示ファイルで見出しとリストを使用すると、Data Engineering Agent に対して指示を整理し、明確にすることができます。
• わかりやすいファイル名を付け、類似した手順を 1 つのファイルにまとめます。Markdown の見出しを使用して、カテゴリや機能ごとにルールを論理的に整理します。
• 指示の競合を避けるため、各指示が適用される特定の条件を明確に定義します。
• プロンプトとワークフローを繰り返し調整します。エージェントの動作は、エージェントのロールアウトとモデルのアップグレードによって時間の経過とともに変化するため、さまざまなプロンプトを使用してルールを繰り返しテストし、改善が必要な領域を特定することをおすすめします。ルールファイルは、データ パイプラインの変更と同期させてください。

次の例は、Data Engineering Agent を効果的に使用するためのベスト プラクティスを活用する、GEMINI.md という名前のエージェント指示ファイルを示しています。

  ### Naming Conventions

  * Datasets: [business_domain]_[use_case] (e.g., ecommerce_sales)

  * Tables:
      - Raw/External: raw_[source_name]
      - Staging: stg_[business_entity]
      - Dimension: dim_[dimension_name]
      - Fact: fct_[fact_name]

  * Dataform Folders:
      - sources
      - staging
      - marts
      - dataProducts

  * Views: vw_[view_name]

  * Columns: snake_case (e.g., order_id, customer_name)

  ## Cloud Storage data load
  * When ingesting data from Cloud Storage, create external tables.

  ## Null handling
  * Filter out null id values

  ## String normalization
  * Standardize string columns by converting to lower case

  ## Data Cleaning Guidelines
  @./generic_cleaning.md