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 dalam 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	Deskripsi
connectionUrl	URL Elasticsearch dalam format https://hostname:[port] atau tentukan CloudID jika menggunakan Elastic Cloud.
apiKey	Kunci API yang Dienkode Base64 yang digunakan untuk autentikasi.
index	Indeks Elasticsearch tempat permintaan akan diajukan, Misalnya, my-index.
inputTableSpec	(Opsional) Tabel BigQuery yang akan dibaca untuk disisipkan ke Elasticsearch. Tabel atau kueri harus diberikan. Misalnya, projectId:datasetId.tablename.
query	(Opsional) Kueri SQL untuk menarik data dari BigQuery. Tabel atau kueri harus diberikan.
useLegacySql	(Opsional) Tetapkan ke true untuk menggunakan legacy SQL (hanya berlaku jika menyediakan kueri). Default: false.
batchSize	(Opsional) Ukuran tumpukan dalam jumlah dokumen. Default: 1000.
batchSizeBytes	(Opsional) Ukuran tumpukan dalam jumlah byte. Default: 5242880 (5mb).
maxRetryAttempts	(Opsional) Percobaan ulang maksimum, harus > 0. Default: no retries.
maxRetryDuration	(Opsional) Durasi percobaan ulang maksimum dalam milidetik, harus > 0. Default: no retries.
propertyAsIndex	(Opsional) Properti dalam dokumen yang diindeks yang nilainya akan menentukan metadata _index untuk disertakan dengan dokumen dalam permintaan massal (lebih diutamakan daripada UDF _index). Default: tidak ada.
propertyAsId	(Opsional) Properti dalam dokumen yang diindeks yang nilainya akan menentukan metadata _id untuk disertakan dengan dokumen dalam permintaan massal (lebih diutamakan daripada UDF _id). Default: tidak ada.
javaScriptIndexFnGcsPath	(Opsional) Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang akan menentukan metadata _index yang akan disertakan dengan dokumen dalam permintaan massal. Default: tidak ada.
javaScriptIndexFnName	(Opsional) Nama fungsi UDF JavaScript untuk fungsi yang akan menentukan metadata _index untuk disertakan dengan dokumen dalam permintaan massal. Default: tidak ada.
javaScriptIdFnGcsPath	(Opsional) Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang akan menentukan metadata _id yang akan disertakan dengan dokumen dalam permintaan massal. Default: tidak ada.
javaScriptIdFnName	(Opsional) Nama fungsi UDF JavaScript untuk fungsi yang akan menentukan metadata _id untuk disertakan dengan dokumen dalam permintaan massal. Default: tidak ada.
javaScriptTypeFnGcsPath	(Opsional) Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang akan menentukan metadata _type yang akan disertakan dengan dokumen dalam permintaan massal. Default: tidak ada.
javaScriptTypeFnName	(Opsional) Nama fungsi UDF JavaScript untuk fungsi yang akan menentukan metadata _type untuk disertakan dengan dokumen dalam permintaan massal. Default: tidak ada.
javaScriptIsDeleteFnGcsPath	(Opsional) Jalur Cloud Storage ke sumber UDF JavaScript untuk fungsi yang akan menentukan apakah dokumen harus dihapus, bukan disisipkan atau diperbarui. Fungsi ini harus menampilkan nilai string "true" atau "false". Default: tidak ada.
javaScriptIsDeleteFnName	(Opsional) Nama fungsi UDF JavaScript untuk fungsi yang akan menentukan apakah dokumen harus dihapus, bukan disisipkan atau diperbarui. Fungsi ini harus menampilkan nilai string "true" atau "false". Default: tidak ada.
usePartialUpdate	(Opsional) Apakah akan menggunakan update sebagian (update, bukan membuat atau mengindeks, mengizinkan dokumen parsial) dengan permintaan Elasticsearch. Default: false.
bulkInsertMethod	(Opsional) Apakah akan menggunakan INDEX (indeks, mengizinkan pembaruan dan penyisipan) atau CREATE (membuat, error pada _id duplikat) dengan permintaan massal Elasticsearch. Default: CREATE.

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 yang memiliki dokumen tersebut.

Parameter template:

• javaScriptIndexFnGcsPath: URI Cloud Storage 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 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. Untuk menggunakan fungsi ini, tetapkan mode sisipkan massal ke INDEX dan sediakan fungsi ID dokumen.

Parameter template:

• javaScriptIsDeleteFnGcsPath: URI Cloud Storage 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 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 Job name, masukkan nama pekerjaan yang unik.
3. Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region default-nya adalah us-central1.

   Untuk 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 parameter value 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 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/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 pekerjaan 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 versi terbaru template, yang tersedia di folder induk tidak bertanggal 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 di folder induk bertanggal masing-masing dalam bucket— gs://dataflow-templates-REGION_NAME/
  Perhatian: Versi terbaru template mungkin diperbarui dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru agar perubahan yang dapat menyebabkan gangguan ini tidak 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 HTTP POST. Untuk 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 pekerjaan 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 versi terbaru template, yang tersedia di folder induk tidak bertanggal 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 di folder induk bertanggal masing-masing dalam bucket— gs://dataflow-templates-REGION_NAME/
  Perhatian: Versi terbaru template mungkin diperbarui dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru agar perubahan yang dapat menyebabkan gangguan ini tidak 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.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 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;

/**
 * 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.
 */
@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"},
    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.",
    })
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.
     *
     *
     * Step #1: Read from BigQuery. If a query is provided then it is used to get the TableRows.
     */
    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)
         */
        .apply(
            TransformTextViaJavascript.newBuilder()
                .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                .setFunctionName(options.getJavascriptTextTransformFunctionName())
                .build())

        /*
         * Step #4: Write converted records to Elasticsearch
         */
        .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.