BigQuery to Elasticsearch テンプレート

BigQuery to Elasticsearch テンプレートは、BigQuery テーブルから Elasticsearch にデータをドキュメントとして取り込むバッチパイプラインです。テンプレートでは、テーブル全体を読み取ることも、クエリを使用して特定のレコードを読み取ることもできます。

パイプラインの要件

ソース BigQuery テーブルが存在すること。
Google Cloud インスタンス上または Elasticsearch バージョン 7.0 以降の Elastic Cloud 上の Elasticsearch ホスト。Dataflow ワーカーマシンからアクセス可能であること。

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

必須パラメータ

connectionUrl : Elasticsearch URL（https://hostname:[port] 形式）。Elastic Cloud を使用する場合は CloudID を指定します（例: https://elasticsearch-host:9200）。
apiKey: 認証に使用される Base64 エンコードの API キー。https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html#security-api-create-api-key-request を参照してください。
index: リクエストが発行されるインデックス（例: my-index）。

オプションパラメータ

inputTableSpec: BigQuery ソーステーブルの仕様（例: bigquery-project:dataset.input_table）。
outputDeadletterTable: さまざまな理由（スキーマの不一致、JSON の形式の誤りなど）で出力テーブルに到達できなかったメッセージが書き込まれるテーブル。このテーブルが存在しない場合は、パイプラインの実行中に作成されます（例: your-project-id:your-dataset.your-table-name）。
query: データを抽出するためにソース上で実行されるクエリ（例: select * from sampledb.sample_table）。
useLegacySql: レガシー SQL を使用するには true に設定します（クエリを提供する場合のみ）。デフォルトは false です。
queryLocation: 基となるテーブルの権限なしで承認済みビューから読み取る場合は必須です（例: US）。
elasticsearchUsername: 認証に使用する Elasticsearch のユーザー名。指定すると、「apiKey」の値は無視されます。
elasticsearchPassword: 認証に使用する Elasticsearch のパスワード。指定すると、「apiKey」の値は無視されます。
batchSize: バッチサイズ（ドキュメント数）。デフォルト値は 1,000 です。
batchSizeBytes: elasticsearch へのメッセージのバッチ挿入に使用されるバッチサイズ（バイト単位）。デフォルト: 「5,242,880（5 MB）」。
maxRetryAttempts: 最大再試行回数。0 より大きな値にする必要があります。デフォルト: 「再試行なし」。
maxRetryDuration: 最大再試行時間（ミリ秒）。0 より大きな値にする必要があります。デフォルト: 「再試行なし」。
propertyAsIndex: インデックスに登録されているドキュメント内のプロパティ。このプロパティの値は、一括リクエストでドキュメントに含まれる「_index」メタデータを指定します（「_index」UDF よりも優先されます）。デフォルト: none
javaScriptIndexFnGcsPath: 一括リクエストでドキュメントに含まれる「_index」メタデータを指定する関数の JavaScript UDF ソースへの Cloud Storage パス。デフォルト: none
javaScriptIndexFnName : 一括リクエストでドキュメントに含まれる「_index」メタデータを指定する関数の UDF JavaScript 関数名。デフォルト: none
propertyAsId: インデックスに登録されているドキュメント内のプロパティ。このプロパティの値は、一括リクエストでドキュメントに含まれる「_id」メタデータを指定します（「_id」UDF よりも優先されます）。デフォルト: none
javaScriptIdFnGcsPath: 一括リクエストでドキュメントに含まれる「_id」メタデータを指定する関数の JavaScript UDF ソースへの Cloud Storage パス。デフォルト: none。
javaScriptIdFnName: 一括リクエストでドキュメントに含まれる「_id」メタデータを指定する関数の UDF JavaScript 関数名。デフォルト: none
javaScriptTypeFnGcsPath: 一括リクエストでドキュメントに含まれる「_type」メタデータを指定する関数の JavaScript UDF ソースへの Cloud Storage パス。デフォルト: none
javaScriptTypeFnName: 一括リクエストでドキュメントに含まれる「_type」メタデータを指定する関数の UDF JavaScript 関数名。デフォルト: none
javaScriptIsDeleteFnGcsPath: ドキュメントを挿入や更新ではなく削除するかどうかを決定する関数の JavaScript UDF ソースへの Cloud Storage パス。この関数は、文字列値「true」または「false」を返す必要があります。デフォルト: none
javaScriptIsDeleteFnName: ドキュメントを挿入や更新ではなく削除するかどうかを決定する関数の UDF JavaScript 関数名。この関数は、文字列値「true」または「false」を返す必要があります。デフォルト: none
usePartialUpdate: Elasticsearch リクエストで部分的な更新（作成やインデックス作成ではなく更新、部分的なドキュメントを許可する）を使用するかどうか。デフォルト: 「false」。
bulkInsertMethod: Elasticsearch 一括リクエストで「INDEX」（インデックス作成、upsert を許可）または「CREATE」（作成、duplicate _id でエラー）を使用するかどうか。デフォルト: 「CREATE」。
trustSelfSignedCerts: 自己署名証明書を信頼するかどうか。インストールされた Elasticsearch インスタンスに自己署名証明書が存在する場合があります。SSL 証明書の検証をバイパスするには、この値を True に設定します（デフォルトは False です）。
disableCertificateValidation: 「true」の場合に、自己署名 SSL 証明書を信頼します。Elasticsearch インスタンスには自己署名証明書が存在する場合があります。証明書の検証をバイパスするには、このパラメータを「true」に設定します。デフォルト: false。
apiKeyKMSEncryptionKey: API キーを復号するための Cloud KMS 鍵。apiKeySource が KMS に設定されている場合は、このパラメータを指定する必要があります。このパラメータを指定する場合は、apiKey 文字列を暗号化して渡す必要があります。KMS API 暗号化エンドポイントを使用してパラメータを暗号化します。鍵は、projects/{gcp_project}/locations/{key_region}/keyRings/{key_ring}/cryptoKeys/{kms_key_name} の形式にする必要があります。https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt をご覧ください（例: projects/your-project-id/locations/global/keyRings/your-keyring/cryptoKeys/your-key-name）。
apiKeySecretId: apiKey の Secret Manager シークレット ID。apiKeySource が SECRET_MANAGER に設定されている場合は、このパラメータを指定する必要があります。projects/{project}/secrets/{secret}/versions/{secret_version} の形式にする必要があります（例: projects/your-project-id/secrets/your-secret/versions/your-secret-version）。
apiKeySource: API キーのソース。PLAINTEXT、KMS、SECRET_MANAGER のいずれかです。Secret Manager または KMS を使用する場合は、このパラメータを指定する必要があります。apiKeySource が KMS に設定されている場合は、apiKeyKMSEncryptionKey と暗号化された apiKey を指定する必要があります。apiKeySource が SECRET_MANAGER に設定されている場合は、apiKeySecretId を指定する必要があります。apiKeySource が PLAINTEXT に設定されている場合は、apiKey を指定する必要があります。デフォルト: PLAINTEXT。
javascriptTextTransformGcsPath: ユーザー定義関数を含む JavaScript コードの Cloud Storage パスパターン（例: gs://your-bucket/your-function.js）。
javascriptTextTransformFunctionName: JavaScript ファイルから呼び出す関数の名前。英字、数字、アンダースコアのみ使用できます（例: transform、transform_udf1）。

ユーザー定義関数

次のように、このテンプレートでは、パイプライン内の複数のポイントでユーザー定義関数（UDF）をサポートしています。詳細については、Dataflow テンプレートのユーザー定義関数を作成するをご覧ください。

インデックス関数

ドキュメントが属するインデックスを返します。

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

javaScriptIndexFnGcsPath: JavaScript ファイルの Cloud Storage URI。
javaScriptIndexFnName: JavaScript 関数の名前。

関数の仕様:

入力: JSON 文字列としてシリアル化された Elasticsearch ドキュメント。
出力: ドキュメントの _index メタデータフィールドの値。

ドキュメント ID 関数

ドキュメント ID を返します。

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

javaScriptIdFnGcsPath: JavaScript ファイルの Cloud Storage URI。
javaScriptIdFnName: JavaScript 関数の名前。

関数の仕様:

入力: JSON 文字列としてシリアル化された Elasticsearch ドキュメント。
出力: ドキュメントの _id メタデータフィールドの値。

ドキュメント削除関数

ドキュメントを削除するかどうかを指定します。この関数を使用するには、一括挿入モードを INDEX に設定し、ドキュメント ID 関数を指定します。

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

javaScriptIsDeleteFnGcsPath: JavaScript ファイルの Cloud Storage URI。
javaScriptIsDeleteFnName: JavaScript 関数の名前。

関数の仕様:

入力: JSON 文字列としてシリアル化された Elasticsearch ドキュメント。
出力: ドキュメントを削除する場合は文字列 "true" を、ドキュメントをアップサートする場合は "false" を返します。

マッピングタイプ関数

ドキュメントのマッピングタイプを返します。

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

javaScriptTypeFnGcsPath: JavaScript ファイルの Cloud Storage URI。
javaScriptTypeFnName: JavaScript 関数の名前。

関数の仕様:

入力: JSON 文字列としてシリアル化された Elasticsearch ドキュメント。
出力: ドキュメントの _type メタデータフィールドの値。

テンプレートを実行する

コンソール

Dataflow の [テンプレートからジョブを作成] ページに移動します。

[テンプレートからジョブを作成] に移動

[ジョブ名] フィールドに、固有のジョブ名を入力します。
（省略可）[リージョンエンドポイント] で、プルダウンメニューから値を選択します。デフォルトのリージョンは us-central1 です。
Dataflow ジョブを実行できるリージョンのリストについては、Dataflow のロケーションをご覧ください。
[Dataflow テンプレート] プルダウンメニューから、[ the BigQuery to Elasticsearch template] を選択します。
表示されたパラメータフィールドに、パラメータ値を入力します。
[ジョブを実行] をクリックします。

gcloud

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

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/BigQuery_to_Elasticsearch \
    --parameters \
inputTableSpec=INPUT_TABLE_SPEC,\
connectionUrl=CONNECTION_URL,\
apiKey=APIKEY,\
index=INDEX

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

PROJECT_ID: Dataflow ジョブを実行する Google Cloud プロジェクトの ID
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/）にあります。
注: 最新のテンプレートでは、互換性のない変更が行われている場合があります。こうした互換性のない変更が本番環境のワークフローに影響しないように、本番環境では最新の日付付き親フォルダに保存されているテンプレートを使用する必要があります。
INPUT_TABLE_SPEC: BigQuery テーブル名。
CONNECTION_URL: Elasticsearch の URL。
APIKEY: 認証用に Base64 でエンコードされた API キー。
INDEX: Elasticsearch インデックス。

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": {
          "inputTableSpec": "INPUT_TABLE_SPEC",
          "connectionUrl": "CONNECTION_URL",
          "apiKey": "APIKEY",
          "index": "INDEX"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/BigQuery_to_Elasticsearch",
   }
}

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

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/）にあります。
注: 最新のテンプレートでは、互換性のない変更が行われている場合があります。こうした互換性のない変更が本番環境のワークフローに影響しないように、本番環境では最新の日付付き親フォルダに保存されているテンプレートを使用する必要があります。
INPUT_TABLE_SPEC: BigQuery テーブル名。
CONNECTION_URL: Elasticsearch の URL。
APIKEY: 認証用に Base64 でエンコードされた API キー。
INDEX: Elasticsearch インデックス。

テンプレートのソースコード

Java

/*
 * Copyright (C) 2021 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.elasticsearch.templates;

import com.google.cloud.teleport.metadata.MultiTemplate;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.elasticsearch.options.BigQueryToElasticsearchOptions;
import com.google.cloud.teleport.v2.elasticsearch.transforms.WriteToElasticsearch;
import com.google.cloud.teleport.v2.transforms.BigQueryConverters.ReadBigQueryTableRows;
import com.google.cloud.teleport.v2.transforms.BigQueryConverters.TableRowToJsonFn;
import com.google.cloud.teleport.v2.transforms.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.v2.transforms.PythonExternalTextTransformer;
import com.google.common.base.Strings;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;

/**
 * The {@link BigQueryToElasticsearch} pipeline exports data from a BigQuery table to Elasticsearch.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-elasticsearch/README_BigQuery_to_Elasticsearch.md">README</a>
 * for instructions on how to use or modify this template.
 */
@MultiTemplate({
  @Template(
      name = "BigQuery_to_Elasticsearch",
      category = TemplateCategory.BATCH,
      displayName = "BigQuery to Elasticsearch",
      description =
          "The BigQuery to Elasticsearch template is a batch pipeline that ingests data from a BigQuery table into Elasticsearch as documents. "
              + "The template can either read the entire table or read specific records using a supplied query.",
      optionsClass = BigQueryToElasticsearchOptions.class,
      skipOptions = {
        "javascriptTextTransformReloadIntervalMinutes",
        "pythonExternalTextTransformGcsPath",
        "pythonExternalTextTransformFunctionName"
      },
      flexContainerName = "bigquery-to-elasticsearch",
      documentation =
          "https://cloud.google.com/dataflow/docs/guides/templates/provided/bigquery-to-elasticsearch",
      contactInformation = "https://cloud.google.com/support",
      preview = true,
      requirements = {
        "The source BigQuery table must exist.",
        "A Elasticsearch host on a Google Cloud instance or on Elastic Cloud with Elasticsearch version 7.0 or above and should be accessible from the Dataflow worker machines.",
      }),
  @Template(
      name = "BigQuery_to_Elasticsearch_Xlang",
      category = TemplateCategory.BATCH,
      displayName = "BigQuery to Elasticsearch with Python UDFs",
      type = Template.TemplateType.XLANG,
      description =
          "The BigQuery to Elasticsearch template is a batch pipeline that ingests data from a BigQuery table into Elasticsearch as documents. "
              + "The template can either read the entire table or read specific records using a supplied query.",
      optionsClass = BigQueryToElasticsearchOptions.class,
      skipOptions = {
        "javascriptTextTransformReloadIntervalMinutes",
        "javascriptTextTransformGcsPath",
        "javascriptTextTransformFunctionName"
      },
      flexContainerName = "bigquery-to-elasticsearch-xlang",
      documentation =
          "https://cloud.google.com/dataflow/docs/guides/templates/provided/bigquery-to-elasticsearch",
      contactInformation = "https://cloud.google.com/support",
      preview = true,
      requirements = {
        "The source BigQuery table must exist.",
        "A Elasticsearch host on a Google Cloud instance or on Elastic Cloud with Elasticsearch version 7.0 or above and should be accessible from the Dataflow worker machines.",
      })
})
public class BigQueryToElasticsearch {
  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    BigQueryToElasticsearchOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(BigQueryToElasticsearchOptions.class);

    run(options);
  }

  /**
   * Runs the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  private static PipelineResult run(BigQueryToElasticsearchOptions options) {

    // Create the pipeline.
    Pipeline pipeline = Pipeline.create(options);
    /*
     * Steps: 1) Read records from BigQuery via BigQueryIO.
     *        2) Create json string from Table Row.
     *        3) Write records to Elasticsearch.
     */

    boolean useJavascriptUdf = !Strings.isNullOrEmpty(options.getJavascriptTextTransformGcsPath());
    boolean usePythonUdf = !Strings.isNullOrEmpty(options.getPythonExternalTextTransformGcsPath());
    if (useJavascriptUdf && usePythonUdf) {
      throw new IllegalArgumentException(
          "Either javascript or Python gcs path must be provided, but not both.");
    }

    /*
     * Step #1: Read from BigQuery. If a query is provided then it is used to get the TableRows.
     */
    PCollection<String> readJsonDocuments =
        pipeline
            .apply(
                "ReadFromBigQuery",
                ReadBigQueryTableRows.newBuilder()
                    .setOptions(options.as(BigQueryToElasticsearchOptions.class))
                    .build())

            /*
             * Step #2: Convert table rows to JSON documents.
             */
            .apply("TableRowsToJsonDocument", ParDo.of(new TableRowToJsonFn()));

    /*
     * Step #3: Apply UDF functions (if specified)
     */
    PCollection<String> udfOut;
    if (usePythonUdf) {
      udfOut =
          readJsonDocuments
              .apply(
                  "MapToRecord",
                  PythonExternalTextTransformer.FailsafeRowPythonExternalUdf
                      .stringMappingFunction())
              .setRowSchema(PythonExternalTextTransformer.FailsafeRowPythonExternalUdf.ROW_SCHEMA)
              .apply(
                  "InvokeUDF",
                  PythonExternalTextTransformer.FailsafePythonExternalUdf.newBuilder()
                      .setFileSystemPath(options.getPythonExternalTextTransformGcsPath())
                      .setFunctionName(options.getPythonExternalTextTransformFunctionName())
                      .build())
              .apply(
                  "MapToStringElements",
                  ParDo.of(new PythonExternalTextTransformer.RowToStringElementFn()));
    } else {
      udfOut =
          readJsonDocuments.apply(
              TransformTextViaJavascript.newBuilder()
                  .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                  .setFunctionName(options.getJavascriptTextTransformFunctionName())
                  .build());
    }

    /*
     * Step #4: Write converted records to Elasticsearch
     */
    udfOut.apply(
        "WriteToElasticsearch",
        WriteToElasticsearch.newBuilder()
            .setUserAgent("dataflow-bigquery-to-elasticsearch-template/v2")
            .setOptions(options.as(BigQueryToElasticsearchOptions.class))
            .build());

    return pipeline.run();
  }
}

次のステップ

Dataflow テンプレートについて学習する。
Google 提供のテンプレートのリストを確認する。

BigQuery to Elasticsearch テンプレート

パイプラインの要件

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

必須パラメータ

オプション パラメータ

ユーザー定義関数

インデックス関数

ドキュメント ID 関数

ドキュメント削除関数

マッピング タイプ関数

テンプレートを実行する

コンソール

gcloud

API

テンプレートのソースコード

Java

次のステップ

オプションパラメータ

マッピングタイプ関数