Template BigQuery ke Elasticsearch

Beta

Fitur ini tunduk pada "Persyaratan Penawaran Pra-GA" di bagian Persyaratan Layanan Umum dalam Persyaratan Khusus Layanan. Fitur pra-GA tersedia "sebagaimana adanya" dan mungkin memiliki dukungan terbatas. Untuk mengetahui informasi selengkapnya, lihat deskripsi tahap peluncuran.

Template BigQuery ke Elasticsearch adalah pipeline batch yang menyerap data dari tabel BigQuery ke Elasticsearch sebagai dokumen. Template dapat membaca seluruh tabel atau membaca data tertentu menggunakan kueri yang disediakan.

Persyaratan pipeline

• Tabel BigQuery sumber harus ada.
• Host Elasticsearch di instance Google Cloud atau di Elastic Cloud dengan Elasticsearch versi 7.0 atau yang lebih baru. Harus dapat diakses dari mesin pekerja Dataflow.

Parameter template

Parameter yang diperlukan

• connectionUrl : URL Elasticsearch dalam format https://hostname:[port]. Jika menggunakan Elastic Cloud, tentukan CloudID. (Contoh: https://elasticsearch-host:9200).
• apiKey : Kunci API yang dienkode Base64 untuk digunakan dalam autentikasi.
• index : Indeks Elasticsearch tempat permintaan dikeluarkan, seperti my-index. (Contoh: my-index).

Parameter opsional

• inputTableSpec : Tabel BigQuery yang akan dibaca. Format: projectId:datasetId.tablename. Jika Anda menentukan inputTableSpec, template akan membaca data langsung dari penyimpanan BigQuery menggunakan BigQuery Storage Read API (https://cloud.google.com/bigquery/docs/reference/storage). Untuk mengetahui informasi tentang batasan di Storage Read API, lihat https://cloud.google.com/bigquery/docs/reference/storage#limitations. Anda harus menentukan inputTableSpec atau query. Jika Anda menetapkan kedua parameter, template akan menggunakan parameter query. (Contoh: bigquery-project:dataset.input_table).
• outputDeadletterTable : Tabel BigQuery untuk pesan yang gagal mencapai tabel output, dalam format <PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE>. Jika tabel tidak ada, tabel akan dibuat selama eksekusi pipeline. Jika tidak ditentukan, <outputTableSpec>_error_records akan digunakan. (Contoh: project-id-Anda:set-data-Anda.nama-tabel-Anda).
• kueri : Kueri SQL yang akan digunakan untuk membaca data dari BigQuery. Jika set data BigQuery berada dalam project yang berbeda dengan tugas Dataflow, tentukan nama set data lengkap dalam kueri SQL, misalnya: <PROJECT_ID>.<DATASET_NAME>.<TABLE_NAME>. Secara default, parameter query menggunakan GoogleSQL (https://cloud.google.com/bigquery/docs/introduction-sql), kecuali jika useLegacySql adalah true. Anda harus menentukan inputTableSpec atau query. Jika Anda menetapkan kedua parameter, template akan menggunakan parameter query. (Contoh: select * from sampledb.sample_table).
• useLegacySql : Tetapkan ke true untuk menggunakan SQL lama. Parameter ini hanya berlaku saat menggunakan parameter query. Defaultnya adalah: false.
• queryLocation : Diperlukan saat membaca dari tampilan yang diotorisasi tanpa izin tabel yang mendasarinya. (Contoh: AS).
• queryTempDataset : Dengan opsi ini, Anda dapat menetapkan set data yang ada untuk membuat tabel sementara guna menyimpan hasil kueri. (Contoh: temp_dataset).
• elasticsearchUsername : Nama pengguna Elasticsearch yang akan digunakan untuk mengautentikasi. Jika ditentukan, nilai 'apiKey' akan diabaikan.
• elasticsearchPassword : Sandi Elasticsearch yang akan digunakan untuk melakukan autentikasi. Jika ditentukan, nilai 'apiKey' akan diabaikan.
• batchSize : Ukuran batch dalam jumlah dokumen. Default-nya adalah: 1.000.
• batchSizeBytes : Ukuran batch dalam jumlah byte. Default-nya adalah: 5242880 (5 mb).
• maxRetryAttempts : Jumlah maksimum percobaan ulang. Harus lebih besar dari nol. Setelan default: tidak ada percobaan ulang.
• maxRetryDuration : Durasi percobaan ulang maksimum dalam milidetik. Harus lebih besar dari nol. Setelan default: tidak ada percobaan ulang.
• propertyAsIndex : Properti dalam dokumen yang diindeks yang nilainya menentukan metadata _index yang akan disertakan dengan dokumen dalam permintaan massal. Lebih diutamakan daripada UDF _index. Setelan defaultnya adalah: none.
• javaScriptIndexFnGcsPath : Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan metadata _index yang akan disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah: none.
• javaScriptIndexFnName : Nama fungsi JavaScript UDF yang menentukan metadata _index untuk disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah: none.
• propertyAsId : Properti dalam dokumen yang diindeks yang nilainya menentukan metadata _id yang akan disertakan dengan dokumen dalam permintaan massal. Lebih diutamakan daripada UDF _id. Setelan defaultnya adalah: none.
• javaScriptIdFnGcsPath : Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan metadata _id yang akan disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah: none.
• javaScriptIdFnName : Nama fungsi JavaScript UDF yang menentukan metadata _id yang akan disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah: none.
• javaScriptTypeFnGcsPath : Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan metadata _type yang akan disertakan dengan dokumen dalam permintaan massal. Default: none.
• javaScriptTypeFnName : Nama fungsi JavaScript UDF yang menentukan metadata _type yang akan disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah: none.
• javaScriptIsDeleteFnGcsPath : Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan apakah akan menghapus dokumen, bukan menyisipkan atau memperbaruinya. Fungsi ini menampilkan nilai string true atau false. Setelan defaultnya adalah: none.
• javaScriptIsDeleteFnName : Nama fungsi JavaScript UDF yang menentukan apakah akan menghapus dokumen, bukan menyisipkan atau memperbaruinya. Fungsi ini menampilkan nilai string true atau false. Setelan defaultnya adalah: none.
• usePartialUpdate : Apakah akan menggunakan pembaruan sebagian (memperbarui, bukan membuat atau mengindeks, yang memungkinkan dokumen sebagian) dengan permintaan Elasticsearch. Defaultnya adalah: false.
• bulkInsertMethod : Apakah akan menggunakan INDEX (indeks, memungkinkan upsert) atau CREATE (buat, error pada _id duplikat) dengan permintaan massal Elasticsearch. Defaultnya adalah: CREATE.
• trustSelfSignedCerts : Apakah akan memercayai sertifikat yang ditandatangani sendiri atau tidak. Instance Elasticsearch yang diinstal mungkin memiliki sertifikat yang ditandatangani sendiri. Aktifkan ini ke Benar untuk mengabaikan validasi pada sertifikat SSL. (defaultnya adalah False).
• disableCertificateValidation : Jika 'true', percayai sertifikat SSL yang ditandatangani sendiri. Instance Elasticsearch mungkin memiliki sertifikat yang ditandatangani sendiri. Untuk mengabaikan validasi sertifikat, tetapkan parameter ini ke 'true'. Default: false.
• apiKeyKMSEncryptionKey : Kunci Cloud KMS untuk mendekripsi kunci API. Parameter ini harus diberikan jika apiKeySource ditetapkan ke KMS. Jika parameter ini diberikan, string apiKey harus diteruskan dalam terenkripsi. Mengenkripsi parameter menggunakan endpoint enkripsi KMS API. Kunci harus dalam format projects/{gcp_project}/locations/{key_region}/keyRings/{key_ring}/cryptoKeys/{kms_key_name}. Lihat: https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt (Contoh: projects/your-project-id/locations/global/keyRings/your-keyring/cryptoKeys/your-key-name).
• apiKeySecretId : ID secret Secret Manager untuk apiKey. Parameter ini harus diberikan jika apiKeySource ditetapkan ke SECRET_MANAGER. Harus dalam format projects/{project}/secrets/{secret}/versions/{secret_version}. (Contoh: projects/your-project-id/secrets/your-secret/versions/your-secret-version).
• apiKeySource : Sumber kunci API. Salah satu dari PLAINTEXT, KMS, atau SECRET_MANAGER. Parameter ini harus diberikan jika pengelola rahasia atau KMS digunakan. Jika apiKeySource ditetapkan ke KMS, apiKeyKMSEncryptionKey dan apiKey terenkripsi harus diberikan. Jika apiKeySource ditetapkan ke SECRET_MANAGER, apiKeySecretId harus diberikan. Jika apiKeySource ditetapkan ke PLAINTEXT, apiKey harus diberikan. Setelan defaultnya adalah: PLAINTEXT.
• socketTimeout : Jika ditetapkan, akan menimpa waktu tunggu percobaan ulang maksimum default dan waktu tunggu soket default (30.000 md) di Elastic RestClient.
• javascriptTextTransformGcsPath : URI Cloud Storage file .js yang menentukan fungsi yang ditentukan pengguna (UDF) JavaScript yang akan digunakan. (Contoh: gs://my-bucket/my-udfs/my_file.js).
• javascriptTextTransformFunctionName : Nama fungsi yang ditentukan pengguna (UDF) JavaScript yang akan digunakan. Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, nama fungsinya adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples).

Fungsi yang ditentukan pengguna (UDF)

Template ini mendukung fungsi yang ditentukan pengguna (UDF) di beberapa titik dalam pipeline, yang dijelaskan di bawah. Untuk informasi selengkapnya, lihat Membuat fungsi yang ditentukan pengguna untuk template Dataflow.

Fungsi indeks

Menampilkan indeks tempat dokumen berada.

Parameter template:

• javaScriptIndexFnGcsPath: URI Cloud Storage dari file JavaScript.
• javaScriptIndexFnName: nama fungsi JavaScript.

Spesifikasi fungsi:

• Input: dokumen Elasticsearch, yang diserialisasi sebagai string JSON.
• Output: nilai kolom metadata _index dokumen.

Fungsi ID dokumen

Menampilkan ID dokumen.

Parameter template:

• javaScriptIdFnGcsPath: URI Cloud Storage dari file JavaScript.
• javaScriptIdFnName: nama fungsi JavaScript.

Spesifikasi fungsi:

• Input: dokumen Elasticsearch, yang diserialisasi sebagai string JSON.
• Output: nilai kolom metadata _id dokumen.

Fungsi penghapusan dokumen

Menentukan apakah akan menghapus dokumen atau tidak. Untuk menggunakan fungsi ini, tetapkan mode penyisipan massal ke INDEX dan berikan fungsi ID dokumen.

Parameter template:

• javaScriptIsDeleteFnGcsPath: URI Cloud Storage dari file JavaScript.
• javaScriptIsDeleteFnName: nama fungsi JavaScript.

Spesifikasi fungsi:

• Input: dokumen Elasticsearch, yang diserialisasi sebagai string JSON.
• Output: menampilkan string "true" untuk menghapus dokumen, atau "false" untuk memperbarui dan menyisipkan dokumen.

Fungsi jenis pemetaan

Menampilkan jenis pemetaan dokumen.

Perhatian: Jenis pemetaan telah dihapus di Elasticsearch 8.0.

Parameter template:

• javaScriptTypeFnGcsPath: URI Cloud Storage dari file JavaScript.
• javaScriptTypeFnName: nama fungsi JavaScript.

Spesifikasi fungsi:

• Input: dokumen Elasticsearch, yang diserialisasi sebagai string JSON.
• Output: nilai kolom metadata _type dokumen.

Menjalankan template

Konsol

1. Buka halaman Create job from template Dataflow.
Buka Buat tugas dari template
2. Di kolom Nama tugas, masukkan nama tugas yang unik.
3. Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region defaultnya adalah us-central1.

   Untuk mengetahui daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.

4. Dari menu drop-down Dataflow template, pilih the BigQuery to Elasticsearch template.
5. Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
6. Klik Run job.

gcloud

Catatan: Untuk menggunakan Google Cloud CLI guna menjalankan template flex, Anda harus memiliki Google Cloud CLI versi 284.0.0, , atau yang lebih baru.

Di shell atau terminal, jalankan template:

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

Ganti kode berikut:

• PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
• JOB_NAME: nama tugas unik pilihan Anda
• REGION_NAME: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
• VERSION: versi template yang ingin Anda gunakan

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan template versi terbaru, yang tersedia di folder induk tanpa tanggal di bucket—gs://dataflow-templates-REGION_NAME/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi template tertentu, yang dapat ditemukan bertingkat dalam folder induk bertanggal masing-masing di bucket—gs://dataflow-templates-REGION_NAME/
  Perhatian: Template versi terbaru mungkin diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru untuk mencegah perubahan yang menyebabkan error ini memengaruhi alur kerja produksi Anda.
• INPUT_TABLE_SPEC: nama tabel BigQuery Anda.
• CONNECTION_URL: URL Elasticsearch Anda.
• APIKEY: kunci API Anda yang dienkode base64 untuk autentikasi.
• INDEX: indeks Elasticsearch Anda.

API

Untuk menjalankan template menggunakan REST API, kirim permintaan POST HTTP. Untuk mengetahui informasi selengkapnya tentang API dan cakupan otorisasinya, lihat 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",
   }
}

Ganti kode berikut:

• PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
• JOB_NAME: nama tugas unik pilihan Anda
• LOCATION: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
• VERSION: versi template yang ingin Anda gunakan

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan template versi terbaru, yang tersedia di folder induk tanpa tanggal di bucket—gs://dataflow-templates-REGION_NAME/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi template tertentu, yang dapat ditemukan bertingkat dalam folder induk bertanggal masing-masing di bucket—gs://dataflow-templates-REGION_NAME/
  Perhatian: Template versi terbaru mungkin diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru untuk mencegah perubahan yang menyebabkan error ini memengaruhi alur kerja produksi Anda.
• INPUT_TABLE_SPEC: nama tabel BigQuery Anda.
• CONNECTION_URL: URL Elasticsearch Anda.
• APIKEY: kunci API Anda yang dienkode base64 untuk autentikasi.
• INDEX: indeks Elasticsearch Anda.

Kode sumber template

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();
  }
}

Langkah selanjutnya

• Pelajari template Dataflow.
• Lihat daftar template yang disediakan Google.