Plantilla de Cloud Storage a Elasticsearch

La plantilla de Cloud Storage a Elasticsearch es una canalización por lotes que lee datos de archivos CSV almacenados en un bucket de Cloud Storage y los escribe en Elasticsearch como documentos de JSON.

Requisitos de la canalización

El bucket de Cloud Storage debe existir.
Debe existir un host de Elasticsearch en una instancia de Google Cloud o en Elasticsearch Cloud al que se pueda acceder desde Dataflow.
Debe existir una tabla de BigQuery para el resultado del error.

Esquema CSV

Si los archivos CSV contienen encabezados, establece el parámetro de plantilla containsHeaders en true.

De lo contrario, crea un archivo de esquema JSON que describa los datos. Especifica el URI de Cloud Storage del archivo de esquema en el parámetro de plantilla jsonSchemaPath. En el siguiente ejemplo, se muestra un esquema JSON:

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

Como alternativa, puedes proporcionar una función definida por el usuario (UDF) que analice el texto CSV y genere documentos de Elasticsearch.

Parámetros de la plantilla

Parámetros obligatorios

deadletterTable: Los mensajes no llegaron al destino por todos los tipos de motivos (p. ej., esquema no coincidente, json con formato incorrecto) se escriben en esta tabla. (Ejemplo: your-project:your-dataset.your-table-name).
inputFileSpec: Patrón del archivo de Cloud Storage desde el que se leerá. P. ej.: gs://tu-bucket/path/*.csv.
connectionUrl : URL de Elasticsearch en el formato “https://hostname:[port]”, o especifica el CloudID si usas Elastic Cloud (ejemplo: https://elasticsearch-host:9200).
apiKey: clave de API codificada en Base64 que se usa para la autenticación. Consulta lo siguiende: https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html#security-api-create-api-key-request.
index: el índice hacia el cual se emitirán las solicitudes (ejemplo: my-index).

Parámetros opcionales

inputFormat: Formato de archivo de entrada. La opción predeterminada es CSV.
containsHeaders: Los archivos CSV de entrada contienen un registro de encabezado (verdadero/falso) Solo se requiere si se leen los archivos CSV. La configuración predeterminada es "false".
delimiter: El delimitador de columnas de los archivos de texto de entrada. Predeterminado: usa el delimitador proporcionado en csvFormat (ejemplo: ,).
csvFormat: Especificación de formato CSV para usar en el análisis de registros. El valor predeterminado es “default”. Consulta https://commons.apache.org/proper/commons-csv/apidocs/org/apache/commons/csv/CSVFormat.html para obtener más detalles. Debe coincidir exactamente con los nombres de formato que se encuentran en: https://commons.apache.org/proper/commons-csv/apidocs/org/apache/commons/csv/CSVFormat.Predefined.html.
jsonSchemaPath: La ruta al esquema JSON. Valor predeterminado: nulo. (Ejemplo: gs://path/to/schema).
largeNumFiles: Configurado como verdadero si el número de archivos está en las decenas de miles. La configuración predeterminada es "false".
csvFileEncoding: Formato de codificación de caracteres de archivo CSV. Los valores permitidos son US-ASCII, ISO-8859-1, UTF-8, UTF-16. La configuración predeterminada es UTF-8.
logDetailedCsvConversionErrors: Se configura como verdadero para habilitar el registro de errores detallado cuando el análisis de CSV falla. Ten en cuenta que esto puede exponer datos sensibles en los registros (p. ej., si el archivo CSV contiene contraseñas). Valor predeterminado: false.
elasticsearchUsername: El nombre de usuario de Elasticsearch con el que se autenticará. Si se especifica, se ignora el valor de “apiKey”.
elasticsearchPassword: La contraseña de Elasticsearch con la que se realizará la autenticación. Si se especifica, se ignora el valor de “apiKey”.
batchSize: El tamaño del lote en cantidad de documentos. Valor predeterminado: "1000".
batchSizeBytes: Tamaño del lote en bytes que se usa para la inserción por lotes de mensajes en elasticsearch. Valor predeterminado: '5242880 (5mb)'.
maxRetryAttempts: La cantidad máxima de reintentos debe ser mayor que 0. Valor predeterminado: "'no retries".
maxRetryDuration: La duración máxima del reintento en milisegundos debe ser superior a 0. Valor predeterminado: "'no retries".
propertyAsIndex: Una propiedad del documento que se indexa, cuyo valor especificará los metadatos "_index" para incluir con el documento en la solicitud masiva (tiene prioridad sobre una UDF "_index"). Valor predeterminado: none.
javaScriptIndexFnGcsPath: La ruta de acceso de Cloud Storage a la fuente de UDF de JavaScript para una función que especificará los metadatos “_index” que se incluirán en el documento en la solicitud masiva. Valor predeterminado: none.
javaScriptIndexFnName: El nombre de función de JavaScript de la UDF para la función que especificará los metadatos que se incluirán en el documento en la solicitud masiva. Valor predeterminado: none.
propertyAsId: Una propiedad del documento que se indexa, cuyo valor especificará los metadatos "_id" para incluir con el documento en la solicitud masiva (tiene prioridad sobre una UDF "_id"). Valor predeterminado: none.
javaScriptIdFnGcsPath: La ruta de acceso de Cloud Storage a la fuente de UDF de JavaScript para una función que especificará los metadatos “_id” que se incluirán en el documento en la solicitud masiva. Predeterminado: ninguno.
javaScriptIdFnName: El nombre de función de JavaScript de la UDF para la función que especificará los metadatos que se incluirán en el documento en la solicitud masiva. Valor predeterminado: none.
javaScriptTypeFnGcsPath: La ruta de Cloud Storage a la fuente de UDF de JavaScript para una función que especificará los metadatos a fin de incluir el documento en la solicitud masiva. Valor predeterminado: none.
javaScriptTypeFnName: Nombre de función de JavaScript de la UDF para la función que especificará los metadatos “_type” que se incluirán en el documento en la solicitud masiva. Valor predeterminado: none.
javaScriptIsDeleteFnGcsPath: La ruta de acceso de Cloud Storage a la fuente de UDF de JavaScript para la función que determinará si el documento se debe borrar en lugar de insertar o actualizar. La función debe mostrar el valor de string "true" o "false". Valor predeterminado: none.
javaScriptIsDeleteFnName: El nombre de función de JavaScript de la UDF para la función que determinará si el documento se debe borrar en lugar de insertarse o actualizarse. La función debe mostrar el valor de string "true" o "false". Valor predeterminado: none.
usePartialUpdate: Indica si se deben usar actualizaciones parciales (actualizar en lugar de crear o indexar), que permite documentos parciales con solicitudes de Elasticsearch. Predeterminado: "false".
bulkInsertMethod : Indica si se debe usar “INDEX” (índice, permite inserción) o “CREATE” (creación, errores en duplicate _id) con solicitudes masivas de Elasticsearch. Configuración predeterminada: “CREATE”.
trustSelfSignedCerts : Indica si se debe confiar o no en el certificado autofirmado. Es posible que una instancia de Elasticsearch instalada tenga un certificado autofirmado. Habilítalo como verdadero para omitir la validación en el certificado SSL. (el valor predeterminado es False).
disableCertificateValidation : si es “true”, confía en el certificado SSL autofirmado. Una instancia de Elasticsearch puede tener un certificado SSL autofirmado. Para omitir la validación del certificado, establece este parámetro como “true”. Valor predeterminado: false.
apiKeyKMSEncryptionKey: La clave de Cloud KMS para desencriptar la clave de API. Este parámetro se debe proporcionar si apiKeySource se configura como KMS. Si se proporciona este parámetro, la cadena apiKey debe pasarse encriptada. Encripta parámetros con el extremo de encriptación de la API de KMS. La clave debe tener el formato proyectos/{proyecto_gcp}/ubicaciones/{región-de-la-clave}/llaveros-de-clave/{llavero-de-claves}/cryptoKeys/{nombre_de_la_clave_kms}. Consulta: https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt (Ejemplo: proyectos/id-de-tu-proyecto/ubicaciones/global/llaveros-de-claves/tu-llavero-de-claves/cryptoKeys/nombre-de-tu-clave).
apiKeySecretId : El ID del secreto de Secret Manager para la apiKey. Este parámetro se debe proporcionar si apiKeySource está configurado como SECRET_MANAGER. Debe tener el formato proyectos/{proyecto}/secrets/{secret}/versiones/{versión-del-secret}. (Ejemplo: proyectos/id-de-tu-proyecto/secrets/tu-secret/versiones/versión-de-tu-secret).
apiKeySource : Fuente de la clave de API. Puede ser PLAINTEXT, KMS o SECRET_MANAGER. Este parámetro se debe proporcionar si se usa Secret Manager o KMS. Si apiKeySource se configura como KMS, se deben proporcionar apiKeyKMSEncryptionKey y la apiKey encriptada. Si apiKeySource se configura como SECRET_MANAGER, se debe proporcionar apiKeySecretId. Si apiKeySource se configura como PLAINTEXT, se debe proporcionar apiKey. La configuración predeterminada es: PLAINTEXT.
javascriptTextTransformGcsPath: el patrón de ruta de acceso de Cloud Storage para el código JavaScript que contiene las funciones definidas por el usuario. (Por ejemplo: gs://your-bucket/your-function.js).
javascriptTextTransformFunctionName : El nombre de la función a la que se llamará desde el archivo JavaScript. Usa solo letras, dígitos y guiones bajos. (Ejemplo: “transform” o “transform_udf1”).

Funciones definidas por el usuario

Esta plantilla admite funciones definidas por el usuario (UDF) en varios puntos de la canalización, que se describen a continuación. Para obtener más información, consulta Crea funciones definidas por el usuario para plantillas de Dataflow.

Función de transformación de texto

Transforma los datos de CSV en un documento de Elasticsearch.

Parámetros de la plantilla:

javascriptTextTransformGcsPath: Es el URI de Cloud Storage del archivo JavaScript.
javascriptTextTransformFunctionName: Es el nombre de la función de JavaScript.

Especificación de la función:

Entrada: una sola línea de un archivo CSV de entrada.
Resultado: Un documento JSON en string para insertar en Elasticsearch.

Función de índice

Muestra el índice al que pertenece el documento.

Parámetros de la plantilla:

javaScriptIndexFnGcsPath: Es el URI de Cloud Storage del archivo JavaScript.
javaScriptIndexFnName: Es el nombre de la función de JavaScript.

Especificación de la función:

Entrada: el documento de Elasticsearch, serializado como una string JSON.
Resultado: El valor del campo de metadatos _index del documento.

Función de ID de documento

Muestra el ID del documento.

Parámetros de la plantilla:

javaScriptIdFnGcsPath: Es el URI de Cloud Storage del archivo JavaScript.
javaScriptIdFnName: Es el nombre de la función de JavaScript.

Especificación de la función:

Entrada: el documento de Elasticsearch, serializado como una cadena JSON.
Resultado: El valor del campo de metadatos _id del documento.

Función de eliminación de documentos

Especifica si se debe borrar un documento. Para usar esta función, establece el modo de inserción masiva en INDEX y proporciona una función de ID de documento.

Parámetros de la plantilla:

javaScriptIsDeleteFnGcsPath: Es el URI de Cloud Storage del archivo JavaScript.
javaScriptIsDeleteFnName: Es el nombre de la función de JavaScript.

Especificación de la función:

Entrada: el documento de Elasticsearch, serializado como una cadena JSON.
Resultado: Muestra la cadena "true" para borrar el documento o "false" a fin de actualizar el documento.

Función de tipo de asignación

Muestra el tipo de asignación del documento.

Parámetros de la plantilla:

javaScriptTypeFnGcsPath: Es el URI de Cloud Storage del archivo JavaScript.
javaScriptTypeFnName: Es el nombre de la función de JavaScript.

Especificación de la función:

Entrada: el documento de Elasticsearch, serializado como una cadena JSON.
Resultado: El valor del campo de metadatos _type del documento.

Ejecuta la plantilla

Consola

Ve a la página Crear un trabajo a partir de una plantilla de Dataflow.

Ir a Crear un trabajo a partir de una plantilla

En el campo Nombre del trabajo, ingresa un nombre de trabajo único.
Opcional: Para Extremo regional, selecciona un valor del menú desplegable. La región predeterminada es us-central1.
Para obtener una lista de regiones en las que puedes ejecutar un trabajo de Dataflow, consulta Ubicaciones de Dataflow.
En el menú desplegable Plantilla de Dataflow, selecciona the Cloud Storage to Elasticsearch template.
En los campos de parámetros proporcionados, ingresa los valores de tus parámetros.
Haga clic en Ejecutar trabajo.

gcloud

En tu shell o terminal, ejecuta la plantilla:

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,\

Reemplaza lo siguiente:

PROJECT_ID: El ID del proyecto de Google Cloud en el que deseas ejecutar el trabajo de Dataflow.
JOB_NAME: Es el nombre del trabajo que elijas
VERSION: Es la versión de la plantilla que deseas usar.
Puedes usar los siguientes valores:
- latest para usar la última versión de la plantilla, que está disponible en la carpeta superior non-dated en el bucket gs://dataflow-templates-REGION_NAME/latest/
- el nombre de la versión, como 2023-09-12-00_RC00, para usar una versión específica de la plantilla, que se puede encontrar anidada en la carpeta superior con fecha correspondiente en el bucket gs://dataflow-templates-REGION_NAME/
Precaución: La versión más reciente de las plantillas podría actualizarse con cambios rotundos. Los entornos de producción deben usar plantillas que se conserven en la carpeta superior con la fecha más reciente para evitar que estos cambios rotundos afecten los flujos de trabajo de producción.
REGION_NAME: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
INPUT_FILE_SPEC: Es el patrón de archivo de Cloud Storage.
CONNECTION_URL: Es tu URL de Elasticsearch
APIKEY: Es tu clave de API codificada en Base64 para la autenticación
INDEX: Es el índice de Elasticsearch.
DEADLETTER_TABLE: Es tu tabla de BigQuery.

API

Para ejecutar la plantilla con la API de REST, envía una solicitud HTTP POST. Para obtener más información de la API y sus permisos de autorización, consulta 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",
   }
}

Reemplaza lo siguiente:

PROJECT_ID: El ID del proyecto de Google Cloud en el que deseas ejecutar el trabajo de Dataflow.
JOB_NAME: Es el nombre del trabajo que elijas
VERSION: Es la versión de la plantilla que deseas usar.
Puedes usar los siguientes valores:
- latest para usar la última versión de la plantilla, que está disponible en la carpeta superior non-dated en el bucket gs://dataflow-templates-REGION_NAME/latest/
- el nombre de la versión, como 2023-09-12-00_RC00, para usar una versión específica de la plantilla, que se puede encontrar anidada en la carpeta superior con fecha correspondiente en el bucket gs://dataflow-templates-REGION_NAME/
Precaución: La versión más reciente de las plantillas podría actualizarse con cambios rotundos. Los entornos de producción deben usar plantillas que se conserven en la carpeta superior con la fecha más reciente para evitar que estos cambios rotundos afecten los flujos de trabajo de producción.
LOCATION: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
INPUT_FILE_SPEC: Es el patrón de archivo de Cloud Storage.
CONNECTION_URL: Es tu URL de Elasticsearch
APIKEY: Es tu clave de API codificada en Base64 para la autenticación
INDEX: Es el índice de Elasticsearch.
DEADLETTER_TABLE: Es tu tabla de BigQuery.

Código fuente de la plantilla

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

¿Qué sigue?

Obtén información sobre las plantillas de Dataflow.
Consulta la lista de plantillas que proporciona Google.