Plantilla de BigQuery a Elasticsearch

Beta

Esta función está sujeta a las "Condiciones de las Ofertas de Fase Previa a la DG" de la sección Condiciones Generales del Servicio de las Condiciones Específicas del Servicio. Las funciones de fase previa a la DG están disponibles “tal cual” y podrían tener asistencia limitada. Para obtener más información, consulta las descripciones de la etapa de lanzamiento.

La plantilla de BigQuery a Elasticsearch es una canalización por lotes que transfiere datos de una tabla de BigQuery a Elasticsearch como documentos. La plantilla puede leer toda la tabla o registros específicos mediante una consulta proporcionada.

Requisitos de la canalización

• La tabla de origen de BigQuery debe existir.
• Un host de Elasticsearch en una instancia de Google Cloud o en Elastic Cloud con Elasticsearch versión 7.0 o posterior. Se debe poder acceder desde las máquinas de trabajador de Dataflow.

Parámetros de la plantilla

Parámetros obligatorios

• connectionUrl: Es la URL de Elasticsearch en el formato https://hostname:[port]. Si usas Elastic Cloud, especifica el CloudID. Por ejemplo, https://elasticsearch-host:9200.
• apiKey: Es la clave de API codificada en Base64 que se usará para la autenticación.
• index: El índice de Elasticsearch al que se emiten las solicitudes. Por ejemplo, my-index

Parámetros opcionales

• inputTableSpec: La tabla de BigQuery desde la que se leerá. Si especificas inputTableSpec, la plantilla lee los datos directamente desde el almacenamiento de BigQuery con la API de BigQuery Storage Read (https://cloud.google.com/bigquery/docs/reference/storage). Para obtener información sobre las limitaciones en la API de Storage Read, consulta https://cloud.google.com/bigquery/docs/reference/storage#limitations. Debes especificar inputTableSpec o query. Si estableces ambos parámetros, la plantilla usa el parámetro query. Por ejemplo, <BIGQUERY_PROJECT>:<DATASET_NAME>.<INPUT_TABLE>.
• outputDeadletterTable: La tabla de BigQuery para los mensajes que no llegaron a la tabla de resultados. Si una tabla no existe, se crea durante la ejecución de la canalización. Si no se especifica, se usa <outputTableSpec>_error_records. Por ejemplo, <PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE>.
• query: La consulta en SQL que se usará para leer datos de BigQuery. Si el conjunto de datos de BigQuery se encuentra en un proyecto diferente que el trabajo de Dataflow, especifica el nombre completo del conjunto de datos en la consulta en SQL, por ejemplo: <PROJECT_ID>.<DATASET_NAME>.<TABLE_NAME>. De forma predeterminada, el parámetro query usa GoogleSQL (https://cloud.google.com/bigquery/docs/introduction-sql), a menos que useLegacySql sea true. Debes especificar inputTableSpec o query. Si estableces ambos parámetros, la plantilla usa el parámetro query. Por ejemplo, select * from sampledb.sample_table.
• useLegacySql: Se establece en true para usar SQL heredado. Este parámetro solo se aplica cuando se usa el parámetro query. La configuración predeterminada es false.
• queryLocation: Se necesita cuando se lee desde una vista autorizada sin el permiso de la tabla subyacente. Por ejemplo, US
• elasticsearchUsername: El nombre de usuario de Elasticsearch con el que se realizará la autenticación. 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. La configuración predeterminada es 1000.
• batchSizeBytes: El tamaño del lote en cantidad de bytes. El valor predeterminado es 5242880 (5 MB).
• maxRetryAttempts: Es la cantidad máxima de reintentos. Debe ser mayor que cero. La configuración predeterminada es no retries.
• maxRetryDuration: Es la duración máxima del reintento en milisegundos. Debe ser mayor que cero. La configuración predeterminada es no retries.
• propertyAsIndex: La propiedad del documento que se indexa, cuyo valor especifica metadatos _index que se incluirán con el documento en las solicitudes masivas. Tiene prioridad sobre una UDF _index. La configuración predeterminada es none.
• javaScriptIndexFnGcsPath: La ruta de Cloud Storage a la fuente de UDF de JavaScript para una función que especifica metadatos _index que se incluirán con el documento en solicitudes masivas. La configuración predeterminada es none.
• javaScriptIndexFnName: El nombre de la función de JavaScript de UDF que especifica los metadatos _index que se incluirán con el documento en las solicitudes masivas. La configuración predeterminada es none.
• propertyAsId: Una propiedad del documento que se indexa, cuyo valor especifica metadatos _id que se incluirán con el documento en las solicitudes masivas. Tiene prioridad sobre una UDF _id. La configuración predeterminada es none.
• javaScriptIdFnGcsPath: La ruta de Cloud Storage a la fuente de UDF de JavaScript para la función que especifica metadatos _id que se incluirán con el documento en solicitudes masivas. La configuración predeterminada es none.
• javaScriptIdFnName: El nombre de la función de JavaScript de UDF que especifica los metadatos _id que se incluirán con el documento en las solicitudes masivas. La configuración predeterminada es none.
• javaScriptTypeFnGcsPath: La ruta de Cloud Storage a la fuente de UDF de JavaScript para una función que especifica metadatos _type que se incluirán en documentos en solicitudes masivas. La configuración predeterminada es none.
• javaScriptTypeFnName: El nombre de la función de JavaScript de UDF que especifica los metadatos _type que se incluirán con el documento en las solicitudes masivas. La configuración predeterminada es none.
• javaScriptIsDeleteFnGcsPath: La ruta de acceso de Cloud Storage a la fuente de UDF de JavaScript para la función que determina si se debe borrar el documento en lugar de insertarlo o actualizarlo. La función devuelve un valor de cadena de true o false. La configuración predeterminada es none.
• javaScriptIsDeleteFnName: El nombre de la función de JavaScript de UDF que determina si se borra el documento en lugar de insertarlo o actualizarlo. La función devuelve un valor de cadena de true o false. La configuración predeterminada es none.
• usePartialUpdate: Indica si se deben usar actualizaciones parciales (actualizar en lugar de crear o indexar, que permite documentos parciales), con solicitudes de Elasticsearch. La configuración predeterminada es 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. La configuración predeterminada es 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 en true. La configuración predeterminada es false.
• apiKeyKMSEncryptionKey: La clave de Cloud KMS para desencriptar la clave de API. Este parámetro es obligatorio si apiKeySource se configura como KMS. Si se proporciona este parámetro, pasa una cadena apiKey encriptada. Encripta parámetros con el extremo de encriptación de la API de KMS. Para la clave, usa el formato projects/<PROJECT_ID>/locations/<KEY_REGION>/keyRings/<KEY_RING>/cryptoKeys/<KMS_KEY_NAME>. Consulta: https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt. Por ejemplo, projects/your-project-id/locations/global/keyRings/your-keyring/cryptoKeys/your-key-name.
• apiKeySecretId: El ID del Secret de Secret Manager para la apiKey. Si apiKeySource se establece como SECRET_MANAGER, proporciona este parámetro. Usa el formato projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. For example, projects/your-project-id/secrets/your-secret/versions/your-secret-version`.
• apiKeySource: Es la fuente de la clave de API. Los valores permitidos son PLAINTEXT, KMS y SECRET_MANAGER. Este parámetro es obligatorio cuando usas Secret Manager o KMS. Si apiKeySource se configura como KMS, apiKeyKMSEncryptionKey y la apiKey encriptada se deben proporcionar. 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.
• socketTimeout: Si se establece, reemplaza el tiempo de espera máximo de reintento y el tiempo de espera de socket predeterminados (30000 ms) en Elastic RestClient.
• javascriptTextTransformGcsPath: El URI de Cloud Storage del archivo .js que define la función definida por el usuario (UDF) de JavaScript que se usará. Por ejemplo, gs://my-bucket/my-udfs/my_file.js
• javascriptTextTransformFunctionName: Es el nombre de la función definida por el usuario (UDF) de JavaScript que se usará. Por ejemplo, si el código de tu función de JavaScript es myTransform(inJson) { /*...do stuff...*/ }, el nombre de la función es myTransform. Para ver ejemplos de UDF de JavaScript, consulta Ejemplos de UDF (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples).

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

Precaución: Se quitaron los tipos de asignación en Elasticsearch 8.0.

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

Console

1. 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
2. En el campo Nombre del trabajo, ingresa un nombre de trabajo único.
3. 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.

4. En el menú desplegable Plantilla de Dataflow, selecciona the BigQuery to Elasticsearch template.
5. En los campos de parámetros proporcionados, ingresa los valores de tus parámetros.
6. Haga clic en Ejecutar trabajo.

gcloud

Nota: Para usar Google Cloud CLI a fin de ejecutar plantillas flex, debes tener Google Cloud CLI en la versión 284.0.0 o una versión posterior.

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/BigQuery_to_Elasticsearch \
    --parameters \
inputTableSpec=INPUT_TABLE_SPEC,\
connectionUrl=CONNECTION_URL,\
apiKey=APIKEY,\
index=INDEX

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
• REGION_NAME: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
• 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.
• INPUT_TABLE_SPEC: Es el nombre de la tabla de BigQuery.
• 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.

API

Para ejecutar la plantilla con la API de REST, envía una solicitud POST HTTP. 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": {
          "inputTableSpec": "INPUT_TABLE_SPEC",
          "connectionUrl": "CONNECTION_URL",
          "apiKey": "APIKEY",
          "index": "INDEX"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/BigQuery_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
• LOCATION: La región en la que deseas implementar tu trabajo de Dataflow, por ejemplo, us-central1
• 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.
• INPUT_TABLE_SPEC: Es el nombre de la tabla de BigQuery.
• 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.

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 com.google.cloud.teleport.metadata.MultiTemplate;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.elasticsearch.options.BigQueryToElasticsearchOptions;
import com.google.cloud.teleport.v2.elasticsearch.transforms.WriteToElasticsearch;
import com.google.cloud.teleport.v2.transforms.BigQueryConverters.ReadBigQueryTableRows;
import com.google.cloud.teleport.v2.transforms.BigQueryConverters.TableRowToJsonFn;
import com.google.cloud.teleport.v2.transforms.JavascriptTextTransformer.TransformTextViaJavascript;
import com.google.cloud.teleport.v2.transforms.PythonExternalTextTransformer;
import com.google.common.base.Strings;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;

/**
 * The {@link BigQueryToElasticsearch} pipeline exports data from a BigQuery table to Elasticsearch.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-elasticsearch/README_BigQuery_to_Elasticsearch.md">README</a>
 * for instructions on how to use or modify this template.
 */
@MultiTemplate({
  @Template(
      name = "BigQuery_to_Elasticsearch",
      category = TemplateCategory.BATCH,
      displayName = "BigQuery to Elasticsearch",
      description =
          "The BigQuery to Elasticsearch template is a batch pipeline that ingests data from a BigQuery table into Elasticsearch as documents. "
              + "The template can either read the entire table or read specific records using a supplied query.",
      optionsClass = BigQueryToElasticsearchOptions.class,
      skipOptions = {
        "javascriptTextTransformReloadIntervalMinutes",
        "pythonExternalTextTransformGcsPath",
        "pythonExternalTextTransformFunctionName"
      },
      flexContainerName = "bigquery-to-elasticsearch",
      documentation =
          "https://cloud.google.com/dataflow/docs/guides/templates/provided/bigquery-to-elasticsearch",
      contactInformation = "https://cloud.google.com/support",
      preview = true,
      requirements = {
        "The source BigQuery table must exist.",
        "A Elasticsearch host on a Google Cloud instance or on Elastic Cloud with Elasticsearch version 7.0 or above and should be accessible from the Dataflow worker machines.",
      }),
  @Template(
      name = "BigQuery_to_Elasticsearch_Xlang",
      category = TemplateCategory.BATCH,
      displayName = "BigQuery to Elasticsearch with Python UDFs",
      type = Template.TemplateType.XLANG,
      description =
          "The BigQuery to Elasticsearch template is a batch pipeline that ingests data from a BigQuery table into Elasticsearch as documents. "
              + "The template can either read the entire table or read specific records using a supplied query.",
      optionsClass = BigQueryToElasticsearchOptions.class,
      skipOptions = {
        "javascriptTextTransformReloadIntervalMinutes",
        "javascriptTextTransformGcsPath",
        "javascriptTextTransformFunctionName"
      },
      flexContainerName = "bigquery-to-elasticsearch-xlang",
      documentation =
          "https://cloud.google.com/dataflow/docs/guides/templates/provided/bigquery-to-elasticsearch",
      contactInformation = "https://cloud.google.com/support",
      preview = true,
      requirements = {
        "The source BigQuery table must exist.",
        "A Elasticsearch host on a Google Cloud instance or on Elastic Cloud with Elasticsearch version 7.0 or above and should be accessible from the Dataflow worker machines.",
      })
})
public class BigQueryToElasticsearch {
  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

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

    run(options);
  }

  /**
   * Runs the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  private static PipelineResult run(BigQueryToElasticsearchOptions options) {

    // Create the pipeline.
    Pipeline pipeline = Pipeline.create(options);
    /*
     * Steps: 1) Read records from BigQuery via BigQueryIO.
     *        2) Create json string from Table Row.
     *        3) Write records to Elasticsearch.
     */

    boolean useJavascriptUdf = !Strings.isNullOrEmpty(options.getJavascriptTextTransformGcsPath());
    boolean usePythonUdf = !Strings.isNullOrEmpty(options.getPythonExternalTextTransformGcsPath());
    if (useJavascriptUdf && usePythonUdf) {
      throw new IllegalArgumentException(
          "Either javascript or Python gcs path must be provided, but not both.");
    }

    /*
     * Step #1: Read from BigQuery. If a query is provided then it is used to get the TableRows.
     */
    PCollection<String> readJsonDocuments =
        pipeline
            .apply(
                "ReadFromBigQuery",
                ReadBigQueryTableRows.newBuilder()
                    .setOptions(options.as(BigQueryToElasticsearchOptions.class))
                    .build())

            /*
             * Step #2: Convert table rows to JSON documents.
             */
            .apply("TableRowsToJsonDocument", ParDo.of(new TableRowToJsonFn()));

    /*
     * Step #3: Apply UDF functions (if specified)
     */
    PCollection<String> udfOut;
    if (usePythonUdf) {
      udfOut =
          readJsonDocuments
              .apply(
                  "MapToRecord",
                  PythonExternalTextTransformer.FailsafeRowPythonExternalUdf
                      .stringMappingFunction())
              .setRowSchema(PythonExternalTextTransformer.FailsafeRowPythonExternalUdf.ROW_SCHEMA)
              .apply(
                  "InvokeUDF",
                  PythonExternalTextTransformer.FailsafePythonExternalUdf.newBuilder()
                      .setFileSystemPath(options.getPythonExternalTextTransformGcsPath())
                      .setFunctionName(options.getPythonExternalTextTransformFunctionName())
                      .build())
              .apply(
                  "MapToStringElements",
                  ParDo.of(new PythonExternalTextTransformer.RowToStringElementFn()));
    } else {
      udfOut =
          readJsonDocuments.apply(
              TransformTextViaJavascript.newBuilder()
                  .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                  .setFunctionName(options.getJavascriptTextTransformFunctionName())
                  .build());
    }

    /*
     * Step #4: Write converted records to Elasticsearch
     */
    udfOut.apply(
        "WriteToElasticsearch",
        WriteToElasticsearch.newBuilder()
            .setUserAgent("dataflow-bigquery-to-elasticsearch-template/v2")
            .setOptions(options.as(BigQueryToElasticsearchOptions.class))
            .build());

    return pipeline.run();
  }
}

¿Qué sigue?

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