Cloud Storage Text to BigQuery（Stream）テンプレート

Cloud Storage Text to BigQuery パイプラインは、Cloud Storage に保存されているテキスト ファイルをストリーミングして、指定された JavaScript ユーザー定義関数（UDF）を使用して変換し、結果を BigQuery に追加するストリーミング パイプラインです。

パイプラインは無期限に実行され、ドレインではなくキャンセルによって手動で終了させる必要があります。これは、分割可能な DoFn で、ドレインをサポートして���ない Watch 変換を使用しているためです。

パイプラインの要件

  • BigQuery で出力テーブルのスキーマを記述する JSON ファイルを作成します。

    fields というタイトルになっているトップレベルの JSON 配列があり、その内容が {"name": "COLUMN_NAME", "type": "DATA_TYPE"} のパターンに従っていることを確認します。例:

    {
  "fields": [
    {
      "name": "name",
      "type": "STRING"
    },
    {
      "name": "age",
      "type": "INTEGER"
    }
  ]
}
  • JavaScript（.js）ファイルを作成し、このファイル内に、テキスト行の変換ロジックを提供する UDF 関数を含めます。使用する関数は、JSON 文字列を返します。

    次の例では、CSV ファイルの各行を分割し、値を含む JSON オブジェクトを作成して、JSON 文字列を返します。

    function process(inJson) {
  val = inJson.split(",");

  const obj = {
    "name": val[0],
    "age": parseInt(val[1])
  };
  return JSON.stringify(obj);
}

テンプレートのパラメータ

パラメータ 説明
javascriptTextTransformGcsPath : 使用する JavaScript ユーザー定義関数（UDF）を定義する .js ファイルの Cloud Storage URI（例: gs://my-bucket/my-udfs/my_file.js）。 ）
JSONPath BigQuery スキーマ ファイルの Cloud Storage の場所。JSON 形式で記述します。例: gs://path/to/my/schema.json。
javascriptTextTransformFunctionName 使用する JavaScript ユーザー定義関数（UDF）の名前。 たとえば、JavaScript 関数が myTransform(inJson) { /*...do stuff...*/ } の場合、関数名は myTransform です。JavaScript UDF の例については、UDF の例をご覧ください。
javascriptTextTransformReloadIntervalMinutes 省略可: UDF を再読み込みする頻度を分単位で指定します。値が 0 より大きい場合、Dataflow は Cloud Storage 内の UDF ファイルを定期的にチェックし、ファイルが変更された場合は UDF を再読み込みします。このパラメータを使用すると、パイプラインの実行中にジョブを再起動することなく、UDF を更新できます。値が 0 の場合、UDF の再読み込みは無効になります。デフォルト値は 0 です。
outputTable 完全修飾された BigQuery テーブル。例: my-project:dataset.table
inputFilePattern 処理するテキストの Cloud Storage の場所。例: gs://my-bucket/my-files/text.txt。
bigQueryLoadingTemporaryDirectory BigQuery 読み込みプロセスで使用する一時ディレクトリ。例: gs://my-bucket/my-files/temp_dir
outputDeadletterTable 出力テーブルに到達できなかったメッセージが記載されたテーブル。例: my-project:dataset.my-unprocessed-table存在しない場合は、パイプラインの実行中に作成されます。指定しない場合は、<outputTableSpec>_error_records が代わりに使用されます。

ユーザー定義関数

このテンプレートには、パイプライン要件で説明されているように、入力ファイルを解析する UDF が必要です。このテンプレートでは、各入力ファイルのテキストごとに UDF を呼び出します。UDF の作成の詳細については、Dataflow テンプレートのユーザー定義関数を作成するをご覧ください。

関数の仕様

UDF の仕様は次のとおりです。

  • 入力: 入力ファイルの 1 行のテキスト。
  • 出力: BigQuery 宛先テーブルのスキーマに一致する JSON 文字列。

テンプレートを実行する

コンソール

  1. Dataflow の [テンプレートからジョブを作成] ページに移動します。
    2. [テンプレートからジョブを作成] に移動
  2. [ジョブ名] フィールドに、固有のジョブ名を入力します。
  3. （省略可）[リージョン エンドポイント] で、プルダウン メニューから値を選択します。デフォルトのリージョンは us-central1 です。

    Dataflow ジョブを実行できるリージョンのリストについては、Dataflow のロケーションをご覧ください。

  4. [Dataflow テンプレート] プルダウン メニューから、[ the Cloud Storage Text to BigQuery (Stream) template] を選択します。
  5. 表示されたパラメータ フィールドに、パラメータ値を入力します。
  6. [ジョブを実行] をクリックします。

gcloud

シェルまたはターミナルで、テンプレートを実行します。

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location gs://dataflow-templates-REGION_NAME/VERSION/flex/Stream_GCS_Text_to_BigQuery_Flex \
    --region REGION_NAME \
    --staging-location STAGING_LOCATION \
    --parameters \
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
JSONPath=PATH_TO_BIGQUERY_SCHEMA_JSON,\
inputFilePattern=PATH_TO_TEXT_DATA,\
outputTable=BIGQUERY_TABLE,\
outputDeadletterTable=BIGQUERY_UNPROCESSED_TABLE,\
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS

次のように置き換えます。

  • JOB_NAME: 一意の任意のジョブ名
  • REGION_NAME: Dataflow ジョブをデプロイするリージョン（例: us-central1）
  • VERSION: 使用するテンプレートのバージョン

    使用できる値は次のとおりです。

    • latest: 最新バージョンのテンプレートを使用します。このテンプレートは、バケット内で日付のない親フォルダ（gs://dataflow-templates-REGION_NAME/latest/）にあります。
    • バージョン名（例: 2023-09-12-00_RC00）。特定のバージョンのテンプレートを使用します。このテンプレートは、バケット内で対応する日付の親フォルダ（gs://dataflow-templates-REGION_NAME/）にあります。
  • STAGING_LOCATION: ローカル ファイルをステージングする場所（例: gs://your-bucket/staging）
  • JAVASCRIPT_FUNCTION: 使用する JavaScript ユーザー定義関数（UDF）の名前

    たとえば、JavaScript 関数のコードが myTransform(inJson) { /*...do stuff...*/ } の場合、関数名は myTransform です。JavaScript UDF の例については、UDF の例をご覧ください。

  • PATH_TO_BIGQUERY_SCHEMA_JSON: スキーマ定義を含む JSON ファイルへの Cloud Storage パス
  • PATH_TO_JAVASCRIPT_UDF_FILE: 使用する JavaScript ユーザー定義関数（UDF）を定義する .js ファイルの Cloud Storage URI（例: gs://my-bucket/my-udfs/my_file.js ）
  • PATH_TO_TEXT_DATA: テキスト データセットの Cloud Storage パス
  • BIGQUERY_TABLE: BigQuery テーブル名
  • BIGQUERY_UNPROCESSED_TABLE: 未処理のメッセージ用の BigQuery テーブルの名前
  • PATH_TO_TEMP_DIR_ON_GCS: 一時ディレクトリへの Cloud Storage パス

API

REST API を使用してテンプレートを実行するには、HTTP POST リクエストを送信します。API とその認証スコープの詳細については、projects.templates.launch をご覧ください。

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
       "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION",
       "JSONPath": "PATH_TO_BIGQUERY_SCHEMA_JSON",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "inputFilePattern":"PATH_TO_TEXT_DATA",
       "outputTable":"BIGQUERY_TABLE",
       "outputDeadletterTable":"BIGQUERY_UNPROCESSED_TABLE",
       "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Stream_GCS_Text_to_BigQuery_Flex",
   }
}

次のように置き換えます。

  • PROJECT_ID: Dataflow ジョブを実行する Google Cloud プロジェクトの ID
  • JOB_NAME: 一意の任意のジョブ名
  • LOCATION: Dataflow ジョブをデプロイするリージョン（例: us-central1）
  • VERSION: 使用するテンプレートのバージョン

    使用できる値は次のとおりです。

    • latest: 最新バージョンのテンプレートを使用します。このテンプレートは、バケット内で日付のない親フォルダ（gs://dataflow-templates-REGION_NAME/latest/）にあります。
    • バージョン名（例: 2023-09-12-00_RC00）。特定のバージョンのテンプレートを使用します。このテンプレートは、バケット内で対応する日付の親フォルダ（gs://dataflow-templates-REGION_NAME/）にあります。
  • STAGING_LOCATION: ローカル ファイルをステージングする場所（例: gs://your-bucket/staging）
  • JAVASCRIPT_FUNCTION: 使用する JavaScript ユーザー定義関数（UDF）の名前

    たとえば、JavaScript 関数のコードが myTransform(inJson) { /*...do stuff...*/ } の場合、関数名は myTransform です。JavaScript UDF の例については、UDF の例をご覧ください。

  • PATH_TO_BIGQUERY_SCHEMA_JSON: スキーマ定義を含む JSON ファイルへの Cloud Storage パス
  • PATH_TO_JAVASCRIPT_UDF_FILE: 使用する JavaScript ユーザー定義関数（UDF）を定義する .js ファイルの Cloud Storage URI（例: gs://my-bucket/my-udfs/my_file.js ）
  • PATH_TO_TEXT_DATA: テキスト データセットの Cloud Storage パス
  • BIGQUERY_TABLE: BigQuery テーブル名
  • BIGQUERY_UNPROCESSED_TABLE: 未処理のメッセージ用の BigQuery テーブルの名前
  • PATH_TO_TEMP_DIR_ON_GCS: 一時ディレクトリへの Cloud Storage パス

次のステップ