Template Cloud Storage 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 Cloud Storage ke Elasticsearch adalah pipeline batch yang membaca data dari file CSV yang disimpan di bucket Cloud Storage dan menulis data ke Elasticsearch sebagai dokumen JSON.

Persyaratan pipeline

• Bucket Cloud Storage harus ada.
• Host Elasticsearch di instance Google Cloud atau di Elasticsearch Cloud yang dapat diakses dari Dataflow harus ada.
• Tabel BigQuery untuk output error harus ada.

Skema CSV

Jika file CSV berisi header, tetapkan parameter template containsHeaders ke true.

Jika tidak, buat file skema JSON yang mendeskripsikan data. Tentukan URI Cloud Storage dari file skema di parameter template jsonSchemaPath. Contoh berikut menunjukkan skema JSON:

[{"name":"id", "type":"text"}, {"name":"age", "type":"integer"}]

Atau, Anda dapat menyediakan fungsi yang ditentukan pengguna (UDF) yang mengurai teks CSV dan menghasilkan dokumen Elasticsearch.

Parameter template

Parameter	Deskripsi
inputFileSpec	Pola file Cloud Storage untuk menelusuri file CSV. Contoh: gs://mybucket/test-*.csv.
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 dikeluarkan, seperti my-index.
deadletterTable	Tabel BigQuery Deadsheet yang akan menerima penyisipan yang gagal. Contoh: <your-project>:<your-dataset>.<your-table-name>.
containsHeaders	(Opsional) Boolean untuk menunjukkan apakah header disertakan dalam CSV. false default.
delimiter	(Opsional) Pemisah yang digunakan CSV. Contoh: ,
csvFormat	(Opsional) Format CSV sesuai dengan format CSV Apache Commons. Default: Default.
jsonSchemaPath	(Opsional) Jalur ke skema JSON. Default: null.
largeNumFiles	(Opsional) Tetapkan ke true jika jumlah file puluhan ribu. Default: false.
javascriptTextTransformGcsPath	(Opsional) URI Cloud Storage dari file .js yang menentukan fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan. Misalnya, gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName	(Opsional) Nama fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan. Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, nama fungsi adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF.
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: tidak ada percobaan ulang.
maxRetryDuration	(Opsional) Durasi percobaan ulang maksimum dalam milidetik, harus > 0. Default: tidak ada percobaan ulang.
csvFileEncoding	(Opsional) Encoding file CSV.
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 transformasi teks

Mengubah data CSV menjadi dokumen Elasticsearch.

Parameter template:

• javascriptTextTransformGcsPath: URI Cloud Storage file JavaScript.
• javascriptTextTransformFunctionName: nama fungsi JavaScript.

Spesifikasi fungsi:

• Input: satu baris dari file CSV input.
• Output: dokumen JSON string yang akan disisipkan ke Elasticsearch.

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 Cloud Storage 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/GCS_to_Elasticsearch \
    --parameters \
inputFileSpec=INPUT_FILE_SPEC,\
connectionUrl=CONNECTION_URL,\
apiKey=APIKEY,\
index=INDEX,\
deadletterTable=DEADLETTER_TABLE,\

Ganti kode berikut:

• PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
• JOB_NAME: nama pekerjaan unik pilihan Anda
• 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.
• REGION_NAME: region tempat Anda ingin men-deploy tugas Dataflow, misalnya us-central1
• INPUT_FILE_SPEC: pola file Cloud Storage Anda.
• CONNECTION_URL: URL Elasticsearch Anda.
• APIKEY: kunci API yang dienkode base64 untuk autentikasi.
• INDEX: indeks Elasticsearch Anda.
• DEADLETTER_TABLE: tabel BigQuery 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": {
          "inputFileSpec": "INPUT_FILE_SPEC",
          "connectionUrl": "CONNECTION_URL",
          "apiKey": "APIKEY",
          "index": "INDEX",
          "deadletterTable": "DEADLETTER_TABLE"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/GCS_to_Elasticsearch",
   }
}

Ganti kode berikut:

• PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
• JOB_NAME: nama pekerjaan unik pilihan Anda
• 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.
• LOCATION: region tempat Anda ingin men-deploy tugas Dataflow, misalnya us-central1
• INPUT_FILE_SPEC: pola file Cloud Storage Anda.
• CONNECTION_URL: URL Elasticsearch Anda.
• APIKEY: kunci API yang dienkode base64 untuk autentikasi.
• INDEX: indeks Elasticsearch Anda.
• DEADLETTER_TABLE: tabel BigQuery 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 static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;

import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.coders.FailsafeElementCoder;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.elasticsearch.options.GCSToElasticsearchOptions;
import com.google.cloud.teleport.v2.elasticsearch.transforms.WriteToElasticsearch;
import com.google.cloud.teleport.v2.transforms.CsvConverters;
import com.google.cloud.teleport.v2.transforms.ErrorConverters.WriteStringMessageErrors;
import com.google.cloud.teleport.v2.utils.SchemaUtils;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.CoderRegistry;
import org.apache.beam.sdk.coders.NullableCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.WithTimestamps;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link GCSToElasticsearch} pipeline exports data from one or more CSV files in Cloud Storage
 * to Elasticsearch.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-elasticsearch/README_GCS_to_Elasticsearch.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "GCS_to_Elasticsearch",
    category = TemplateCategory.BATCH,
    displayName = "Cloud Storage to Elasticsearch",
    description = {
      "The Cloud Storage to Elasticsearch template is a batch pipeline that reads data from CSV files stored in a Cloud Storage bucket and writes the data into Elasticsearch as JSON documents.",
      "If the CSV files contain headers, set the <code>containsHeaders</code> template parameter to <code>true</code>.\n"
          + "Otherwise, create a JSON schema file that describes the data. Specify the Cloud Storage URI of the schema file in the jsonSchemaPath template parameter. "
          + "The following example shows a JSON schema:\n"
          + "<code>[{\"name\":\"id\", \"type\":\"text\"}, {\"name\":\"age\", \"type\":\"integer\"}]</code>\n"
          + "Alternatively, you can provide a user-defined function (UDF) that parses the CSV text and outputs Elasticsearch documents."
    },
    optionsClass = GCSToElasticsearchOptions.class,
    skipOptions = {"javascriptTextTransformReloadIntervalMinutes"},
    flexContainerName = "gcs-to-elasticsearch",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-storage-to-elasticsearch",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The Cloud Storage bucket must exist.",
      "A Elasticsearch host on a Google Cloud instance or on Elasticsearch Cloud that is accessible from Dataflow must exist.",
      "A BigQuery table for error output must exist."
    })
public class GCSToElasticsearch {

  /** The tag for the headers of the CSV if required. */
  static final TupleTag<String> CSV_HEADERS = new TupleTag<String>() {};

  /** The tag for the lines of the CSV. */
  static final TupleTag<String> CSV_LINES = new TupleTag<String>() {};

  /** The tag for the dead-letter output of the UDF. */
  static final TupleTag<FailsafeElement<String, String>> PROCESSING_DEADLETTER_OUT =
      new TupleTag<FailsafeElement<String, String>>() {};

  /** The tag for the main output for the UDF. */
  static final TupleTag<FailsafeElement<String, String>> PROCESSING_OUT =
      new TupleTag<FailsafeElement<String, String>>() {};

  /* Logger for class. */
  private static final Logger LOG = LoggerFactory.getLogger(GCSToElasticsearch.class);

  /** String/String Coder for FailsafeElement. */
  private static final FailsafeElementCoder<String, String> FAILSAFE_ELEMENT_CODER =
      FailsafeElementCoder.of(
          NullableCoder.of(StringUtf8Coder.of()), NullableCoder.of(StringUtf8Coder.of()));

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

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

    run(options);
  }

  /**
   * Runs the pipeline to completion with the specified options.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  private static PipelineResult run(GCSToElasticsearchOptions options) {
    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    // Register the coder for pipeline
    CoderRegistry coderRegistry = pipeline.getCoderRegistry();
    coderRegistry.registerCoderForType(
        FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor(), FAILSAFE_ELEMENT_CODER);

    // Throw error if containsHeaders is true and a schema or Udf is also set.
    if (options.getContainsHeaders()) {
      checkArgument(
          options.getJavascriptTextTransformGcsPath() == null
              && options.getJsonSchemaPath() == null,
          "Cannot parse file containing headers with UDF or Json schema.");
    }

    // Throw error if only one retry configuration parameter is set.
    checkArgument(
        (options.getMaxRetryAttempts() == null && options.getMaxRetryDuration() == null)
            || (options.getMaxRetryAttempts() != null && options.getMaxRetryDuration() != null),
        "To specify retry configuration both max attempts and max duration must be set.");

    /*
     * Steps: 1) Read records from CSV(s) via {@link CsvConverters.ReadCsv}.
     *        2) Convert lines to JSON strings via {@link CsvConverters.LineToFailsafeJson}.
     *        3a) Write JSON strings as documents to Elasticsearch via {@link ElasticsearchIO}.
     *        3b) Write elements that failed processing to {@link org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO}.
     */
    PCollectionTuple convertedCsvLines =
        pipeline
            /*
             * Step 1: Read CSV file(s) from Cloud Storage using {@link CsvConverters.ReadCsv}.
             */
            .apply(
                "ReadCsv",
                CsvConverters.ReadCsv.newBuilder()
                    .setCsvFormat(options.getCsvFormat())
                    .setDelimiter(options.getDelimiter())
                    .setHasHeaders(options.getContainsHeaders())
                    .setInputFileSpec(options.getInputFileSpec())
                    .setHeaderTag(CSV_HEADERS)
                    .setLineTag(CSV_LINES)
                    .setFileEncoding(options.getCsvFileEncoding())
                    .build())
            /*
             * Step 2: Convert lines to Elasticsearch document.
             */
            .apply(
                "ConvertLine",
                CsvConverters.LineToFailsafeJson.newBuilder()
                    .setDelimiter(options.getDelimiter())
                    .setUdfFileSystemPath(options.getJavascriptTextTransformGcsPath())
                    .setUdfFunctionName(options.getJavascriptTextTransformFunctionName())
                    .setJsonSchemaPath(options.getJsonSchemaPath())
                    .setHeaderTag(CSV_HEADERS)
                    .setLineTag(CSV_LINES)
                    .setUdfOutputTag(PROCESSING_OUT)
                    .setUdfDeadletterTag(PROCESSING_DEADLETTER_OUT)
                    .build());
    /*
     * Step 3a: Write elements that were successfully processed to Elasticsearch using {@link WriteToElasticsearch}.
     */
    convertedCsvLines
        .get(PROCESSING_OUT)
        .apply(
            "GetJsonDocuments",
            MapElements.into(TypeDescriptors.strings()).via(FailsafeElement::getPayload))
        .apply(
            "WriteToElasticsearch",
            WriteToElasticsearch.newBuilder()
                .setUserAgent("dataflow-gcs-to-elasticsearch-template/v2")
                .setOptions(options.as(GCSToElasticsearchOptions.class))
                .build());

    /*
     * Step 3b: Write elements that failed processing to deadletter table via {@link BigQueryIO}.
     */
    convertedCsvLines
        .get(PROCESSING_DEADLETTER_OUT)
        .apply(
            "AddTimestamps",
            WithTimestamps.of((FailsafeElement<String, String> failures) -> new Instant()))
        .apply(
            "WriteFailedElementsToBigQuery",
            WriteStringMessageErrors.newBuilder()
                .setErrorRecordsTable(options.getDeadletterTable())
                .setErrorRecordsTableSchema(SchemaUtils.DEADLETTER_SCHEMA)
                .build());

    return pipeline.run();
  }
}

Langkah selanjutnya

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