Template BigQuery ke Elasticsearch

Beta

Fitur ini tunduk pada "Persyaratan Penawaran Pra-GA" di bagian Persyaratan Layanan Umum dari 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 ini dapat membaca seluruh tabel atau membaca rekaman tertentu menggunakan kueri yang disediakan.

Persyaratan pipeline

• Tabel BigQuery sumber harus ada.
• Host Elasticsearch di Google Cloud instance 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. Misalnya, https://elasticsearch-host:9200.
• apiKey: Kunci API yang dienkode Base64 untuk digunakan dalam autentikasi.
• index: Indeks Elasticsearch tempat permintaan dikeluarkan. Misalnya, my-index.

Parameter opsional

• inputTableSpec: Tabel BigQuery yang akan dibaca. 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. Misalnya, <BIGQUERY_PROJECT>:<DATASET_NAME>.<INPUT_TABLE>.
• outputDeadletterTable: Tabel BigQuery untuk pesan yang gagal mencapai tabel output. Jika tidak ada, tabel akan dibuat selama eksekusi pipeline. Jika tidak ditentukan, <outputTableSpec>_error_records akan digunakan. Misalnya, <PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE>.
• query: Kueri SQL yang akan digunakan untuk membaca data dari BigQuery. Jika set data BigQuery berada di 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. Misalnya, select * from sampledb.sample_table.
• useLegacySql: Tetapkan ke true untuk menggunakan legacy SQL. Parameter ini hanya berlaku saat menggunakan parameter query. Setelan defaultnya adalah false.
• queryLocation: Diperlukan saat membaca dari tabel virtual yang diizinkan tanpa izin tabel yang mendasarinya. Misalnya, US.
• queryTempDataset: Dengan opsi ini, Anda dapat menetapkan set data yang ada untuk membuat tabel sementara guna menyimpan hasil kueri. Misalnya, temp_dataset.
• KMSEncryptionKey: Jika membaca dari BigQuery menggunakan sumber kueri, gunakan kunci Cloud KMS ini untuk mengenkripsi tabel sementara yang dibuat. Misalnya, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
• elasticsearchUsername: Nama pengguna Elasticsearch untuk autentikasi. Jika ditentukan, nilai apiKey akan diabaikan.
• elasticsearchPassword: Sandi Elasticsearch untuk autentikasi. Jika ditentukan, nilai apiKey akan diabaikan.
• batchSize: Ukuran batch dalam jumlah dokumen. Setelan defaultnya adalah 1000.
• batchSizeBytes: Ukuran batch dalam jumlah byte. Setelan defaultnya adalah 5242880 (5 MB).
• maxRetryAttempts: Jumlah maksimum percobaan ulang. Harus lebih besar dari nol. Setelan defaultnya adalah 5.
• maxRetryDuration: Durasi percobaan ulang maksimum dalam milidetik. Harus lebih besar dari nol. Setelan defaultnya adalah 60000 (1 menit).
• propertyAsIndex: Properti dalam dokumen yang diindeks yang nilainya menentukan metadata _index untuk disertakan dengan dokumen dalam permintaan massal. Diprioritaskan daripada UDF _index. Setelan defaultnya adalah none.
• javaScriptIndexFnGcsPath: Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan metadata _index untuk 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 untuk disertakan dengan dokumen dalam permintaan massal. Diprioritaskan daripada UDF _id. Setelan defaultnya adalah none.
• javaScriptIdFnGcsPath: Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan metadata _id untuk disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah none.
• javaScriptIdFnName: Nama fungsi JavaScript UDF yang menentukan metadata _id untuk disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah none.
• javaScriptTypeFnGcsPath: Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang menentukan metadata _type untuk disertakan dengan dokumen dalam permintaan massal. Setelan defaultnya adalah none.
• javaScriptTypeFnName: Nama fungsi JavaScript UDF yang menentukan metadata _type untuk 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 update sebagian (memperbarui, bukan membuat atau mengindeks, yang memungkinkan dokumen sebagian) dengan permintaan Elasticsearch. Setelan defaultnya adalah false.
• bulkInsertMethod: Apakah akan menggunakan INDEX (indeks, memungkinkan upsert) atau CREATE (membuat, error pada _id duplikat) dengan permintaan massal Elasticsearch. Setelan defaultnya adalah CREATE.
• trustSelfSignedCerts: Apakah akan mempercayai sertifikat yang ditandatangani sendiri atau tidak. Instance Elasticsearch yang diinstal mungkin memiliki sertifikat yang ditandatangani sendiri. Aktifkan parameter ini ke true untuk melewati validasi pada sertifikat SSL. (Setelan defaultnya adalah: false).
• disableCertificateValidation: Jika true, percayai sertifikat SSL yang ditandatangani sendiri. Instance Elasticsearch mungkin memiliki sertifikat yang ditandatangani sendiri. Untuk melewati validasi sertifikat, tetapkan parameter ini ke true. Setelan defaultnya adalah false.
• apiKeyKMSEncryptionKey: Kunci Cloud KMS untuk mendekripsi kunci API. Parameter ini diperlukan jika apiKeySource ditetapkan ke KMS. Jika parameter ini disediakan, teruskan string apiKey yang dienkripsi. Enkripsi parameter menggunakan endpoint enkripsi KMS API. Untuk kunci, gunakan format projects/<PROJECT_ID>/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 Misalnya, projects/your-project-id/locations/global/keyRings/your-keyring/cryptoKeys/your-key-name.
• apiKeySecretId: ID secret Secret Manager untuk apiKey. Jika apiKeySource ditetapkan ke SECRET_MANAGER, berikan parameter ini. Gunakan format projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. For example, projects/your-project-id/secrets/your-secret/versions/your-secret-version`.
• apiKeySource: Sumber kunci API. Nilai yang diizinkan adalah PLAINTEXT, KMS, atau SECRET_MANAGER. Parameter ini diperlukan saat Anda menggunakan Secret Manager atau KMS. Jika apiKeySource ditetapkan ke KMS, apiKeyKMSEncryptionKey dan apiKey yang dienkripsi harus disediakan. Jika apiKeySource ditetapkan ke SECRET_MANAGER, apiKeySecretId harus disediakan. Jika apiKeySource ditetapkan ke PLAINTEXT, apiKey harus disediakan. Setelan defaultnya adalah: PLAINTEXT.
• socketTimeout: Jika ditetapkan, akan menimpa waktu tunggu percobaan ulang maksimum default dan waktu tunggu soket default (30000 md) di Elastic RestClient.
• javascriptTextTransformGcsPath: Cloud Storage URI dari 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...*/ }, maka 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 mengetahui informasi selengkapnya, lihat Membuat fungsi yang ditentukan pengguna (UDF) untuk template Dataflow.

Fungsi indeks

Menampilkan indeks tempat dokumen berada.

Parameter template:

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

Spesifikasi fungsi:

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

Fungsi ID dokumen

Menampilkan ID dokumen.

Parameter template:

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

Spesifikasi fungsi:

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

Fungsi penghapusan dokumen

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

Parameter template:

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

Spesifikasi fungsi:

• Input: dokumen Elasticsearch, diserialisasi sebagai string JSON.
• Output: menampilkan string "true" untuk menghapus dokumen, atau "false" untuk melakukan upsert dokumen.

Fungsi jenis pemetaan

Menampilkan jenis pemetaan dokumen.

Perhatian: Jenis pemetaan dihapus di Elasticsearch 8.0.

Parameter template:

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

Spesifikasi fungsi:

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

Menjalankan template

Konsol

1. Buka halaman Dataflow Membuat tugas dari template.
Buka Membuat 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 default-nya adalah us-central1.

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

4. Dari menu drop-down Template Dataflow, pilih template BigQuery ke Elasticsearch.
5. Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
6. Klik Jalankan tugas.

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 Anda, 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/ \
    --parameters \
inputTableSpec=INPUT_TABLE_SPEC,\
connectionUrl=CONNECTION_URL,\
apiKey=APIKEY,\
index=INDEX

Ganti kode berikut:

• PROJECT_ID: Google Cloud Project ID 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 template versi tertentu, yang dapat ditemukan bertingkat di masing-masing folder induk yang diberi tanggal dalam bucket—gs://dataflow-templates-REGION_NAME/
  Perhatian: Template versi terbaru dapat diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk berisi tanggal terbaru untuk mencegah perubahan yang dapat menyebabkan gangguan ini memengaruhi alur kerja produksi Anda.
• INPUT_TABLE_SPEC: nama tabel BigQuery Anda.
• CONNECTION_URL: URL Elasticsearch Anda.
• APIKEY: kunci API 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/",
   }
}

Ganti kode berikut:

• PROJECT_ID: Google Cloud Project ID 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 template versi tertentu, yang dapat ditemukan bertingkat di masing-masing folder induk yang diberi tanggal dalam bucket—gs://dataflow-templates-REGION_NAME/
  Perhatian: Template versi terbaru dapat diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk berisi tanggal terbaru untuk mencegah perubahan yang dapat menyebabkan gangguan ini memengaruhi alur kerja produksi Anda.
• INPUT_TABLE_SPEC: nama tabel BigQuery Anda.
• CONNECTION_URL: URL Elasticsearch Anda.
• APIKEY: kunci API 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 = "googlecloud-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 = "googlecloud-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 berikutnya

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