Template BigQuery ke Elasticsearch

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 diberikan.

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. Misalnya, https://elasticsearch-host:9200.
• apiKey: Kunci API berenkode Base64 yang akan digunakan untuk autentikasi.
• index: Indeks Elasticsearch tempat permintaan dikeluarkan. Contoh, 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>.
• kueri: 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: Setel ke true untuk menggunakan legacy SQL. Parameter ini hanya berlaku saat menggunakan parameter query. Setelan defaultnya adalah false.
• queryLocation: Diperlukan saat membaca dari tampilan yang diizinkan tanpa izin tabel pokok. Contoh, US.
• queryTempDataset: Dengan opsi ini, Anda dapat menetapkan set data yang ada untuk membuat tabel sementara guna menyimpan hasil kueri. Contoh, temp_dataset.
• KMSEncryptionKey: Jika membaca dari BigQuery menggunakan sumber kueri, gunakan kunci Cloud KMS ini untuk mengenkripsi tabel sementara yang dibuat. Contoh, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
• elasticsearchUsername: Nama pengguna Elasticsearch untuk mengautentikasi. Jika ditentukan, nilai apiKey akan diabaikan.
• elasticsearchPassword: Sandi Elasticsearch untuk melakukan autentikasi. Jika ditentukan, nilai apiKey akan diabaikan.
• batchSize: Ukuran batch dalam jumlah dokumen. Nilai defaultnya adalah 1000.
• batchSizeBytes: Ukuran batch dalam jumlah byte. Nilai defaultnya adalah 5242880 (5 MB).
• maxRetryAttempts: Jumlah maksimum upaya percobaan ulang. Harus lebih besar dari nol. Nilai defaultnya adalah no retries.
• maxRetryDuration: Durasi percobaan ulang maksimum dalam milidetik. Harus lebih besar dari nol. Nilai defaultnya adalah no retries.
• propertyAsIndex: Properti dalam dokumen yang diindeks yang nilainya menentukan metadata _index untuk disertakan dengan dokumen dalam permintaan massal. Lebih diprioritaskan 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 untuk disertakan dengan dokumen dalam permintaan massal. Lebih diprioritaskan 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. 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 dokumen akan dihapus, bukan disisipkan atau diperbarui. Fungsi ini menampilkan nilai string true atau false. Setelan defaultnya adalah none.
• usePartialUpdate: Apakah akan menggunakan update parsial (update, bukan membuat atau mengindeks, yang memungkinkan dokumen parsial) dengan permintaan Elasticsearch. Nilai defaultnya adalah false.
• bulkInsertMethod: Apakah akan menggunakan INDEX (mengindeks, memungkinkan operasi upsert) atau CREATE (membuat, error pada _id duplikat) dengan permintaan massal Elasticsearch. Setelan defaultnya adalah CREATE.
• trustSelfSignedCerts: Apakah akan memercayai sertifikat yang ditandatangani sendiri atau tidak. Instance Elasticsearch yang diinstal mungkin memiliki sertifikat yang ditandatangani sendiri. Setel opsi ini ke benar (true) untuk melewati validasi pada sertifikat SSL. (Default: 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 wajib diisi jika apiKeySource ditetapkan ke KMS. Jika parameter ini diberikan, teruskan string apiKey terenkripsi. 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 terenkripsi harus diberikan. Jika apiKeySource ditetapkan ke SECRET_MANAGER, apiKeySecretId harus diberikan. Jika apiKeySource ditetapkan ke PLAINTEXT, apiKey harus diberikan. Defaultnya adalah: PLAINTEXT.
• socketTimeout: Jika disetel, akan menggantikan waktu tunggu percobaan ulang maksimum default dan waktu tunggu soket default (30000 md) di Elastic RestClient.
• javascriptTextTransformGcsPath: URI Cloud Storage 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 untuk template Dataflow.

Fungsi indeks

Menampilkan indeks tempat dokumen berada.

Parameter template:

• javaScriptIndexFnGcsPath: Cloud Storage URI 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 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 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 meng-upsert dokumen.

Fungsi jenis pemetaan

Menampilkan jenis pemetaan dokumen.

Parameter template:

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

Spesifikasi fungsi:

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

Menjalankan 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

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",
   }
}

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