Halaman ini diterjemahkan oleh Cloud Translation API.

Template Cloud Storage ke Elasticsearch

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 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 yang diperlukan

deadletterTable : Tabel dead-letter BigQuery tempat penyisipan yang gagal akan dikirim. (Contoh: project-anda:set-data-anda.nama-tabel-anda).
inputFileSpec : Pola file Cloud Storage untuk menelusuri file CSV. Contoh: gs://mybucket/test-*.csv.
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

inputFormat : Format file input. Default-nya adalah: CSV.
containsHeaders : File CSV input berisi data header (true/false). Hanya diperlukan jika membaca file CSV. Defaultnya adalah: false.
delimiter : Pemisah kolom file teks input. Default: gunakan pemisah yang disediakan di csvFormat (Contoh: ,).
csvFormat : Spesifikasi format CSV yang akan digunakan untuk mengurai data. Default-nya adalah: Default. Lihat https://commons.apache.org/proper/commons-csv/apidocs/org/apache/commons/csv/CSVFormat.html untuk detail selengkapnya. Harus sama persis dengan nama format yang ditemukan di: https://commons.apache.org/proper/commons-csv/apidocs/org/apache/commons/csv/CSVFormat.Predefined.html.
jsonSchemaPath : Jalur ke skema JSON. Nilai defaultnya adalah: null. (Contoh: gs://path/to/schema).
largeNumFiles : Tetapkan ke benar jika jumlah file mencapai puluhan ribu. Defaultnya adalah: false.
csvFileEncoding : Format encoding karakter file CSV. Nilai yang Diizinkan adalah US-ASCII, ISO-8859-1, UTF-8, dan UTF-16. Setelan defaultnya adalah: UTF-8.
logDetailedCsvConversionErrors : Tetapkan ke true untuk mengaktifkan logging error mendetail saat penguraian CSV gagal. Perhatikan bahwa tindakan ini dapat mengekspos data sensitif dalam log (misalnya, jika file CSV berisi sandi). Default: false.
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 dokumen akan dihapus, bukan disisipkan atau diperbarui. 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 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 yang di-stringifikasi untuk disisipkan ke dalam Elasticsearch.

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.

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

Buka halaman Create job from template Dataflow.

Buka Buat tugas dari template

Di kolom Nama tugas, masukkan nama tugas yang unik.
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.
Dari menu drop-down Dataflow template, pilih the Cloud Storage to Elasticsearch template.
Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
Klik Run job.

gcloud

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/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 tugas unik pilihan Anda
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.
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 Anda yang dienkode base64 untuk autentikasi.
INDEX: indeks Elasticsearch Anda.
DEADLETTER_TABLE: tabel BigQuery 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": {
          "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 tugas unik pilihan Anda
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.
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 Anda 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.MultiTemplate;
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 com.google.common.base.Strings;
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.
 */
@MultiTemplate({
  @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 Javascript user-defined function (UDF) that parses the CSV text and outputs Elasticsearch documents."
      },
      optionsClass = GCSToElasticsearchOptions.class,
      skipOptions = {
        "javascriptTextTransformReloadIntervalMinutes",
        "pythonExternalTextTransformGcsPath",
        "pythonExternalTextTransformFunctionName"
      },
      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."
      }),
  @Template(
      name = "GCS_to_Elasticsearch_Xlang",
      category = TemplateCategory.BATCH,
      displayName = "Cloud Storage to Elasticsearch with Python UDFs",
      type = Template.TemplateType.XLANG,
      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 Python user-defined function (UDF) that parses the CSV text and outputs Elasticsearch documents."
      },
      optionsClass = GCSToElasticsearchOptions.class,
      skipOptions = {
        "javascriptTextTransformGcsPath",
        "javascriptTextTransformFunctionName",
        "javascriptTextTransformReloadIntervalMinutes"
      },
      flexContainerName = "gcs-to-elasticsearch-xlang",
      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
              && options.getPythonExternalTextTransformGcsPath() == 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.");

    // Throw error if both Javascript UDF and Python UDF are set. We can only apply one or the
    // other.
    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.");
    }

    /*
     * 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 readCsvLines =
        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.
     */
    CsvConverters.LineToFailsafeJson.Builder lineToFailsafeJsonBuilder =
        CsvConverters.LineToFailsafeJson.newBuilder()
            .setDelimiter(options.getDelimiter())
            .setJsonSchemaPath(options.getJsonSchemaPath())
            .setHeaderTag(CSV_HEADERS)
            .setLineTag(CSV_LINES)
            .setUdfOutputTag(PROCESSING_OUT)
            .setUdfDeadletterTag(PROCESSING_DEADLETTER_OUT);
    if (options.getPythonExternalTextTransformGcsPath() != null) {
      lineToFailsafeJsonBuilder
          .setPythonUdfFileSystemPath(options.getPythonExternalTextTransformGcsPath())
          .setPythonUdfFunctionName(options.getPythonExternalTextTransformFunctionName());
    } else {
      lineToFailsafeJsonBuilder
          .setJavascriptUdfFileSystemPath(options.getJavascriptTextTransformGcsPath())
          .setJavascriptUdfFunctionName(options.getJavascriptTextTransformFunctionName());
    }
    PCollectionTuple convertedCsvLines =
        readCsvLines.apply("ConvertLine", lineToFailsafeJsonBuilder.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.