Esta página se ha traducido con Cloud Translation API.

Plantilla de BigQuery a Elasticsearch

La plantilla de BigQuery a Elasticsearch es un flujo de procesamiento por lotes que ingiere datos de una tabla de BigQuery en Elasticsearch como documentos. La plantilla puede leer toda la tabla o registros específicos mediante una consulta proporcionada.

Requisitos del flujo de procesamiento

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

Parámetros de plantilla

Parámetros obligatorios

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

Parámetros opcionales

inputTableSpec la tabla de BigQuery de la que se van a leer los datos. Si especificas inputTableSpec, la plantilla lee los datos directamente del almacenamiento de BigQuery mediante la API Storage Read de BigQuery (https://cloud.google.com/bigquery/docs/reference/storage). Para obtener información sobre las limitaciones de la API Storage Read, consulta https://cloud.google.com/bigquery/docs/reference/storage#limitations. Debes especificar inputTableSpec o query. Si defines ambos parámetros, la plantilla usará el parámetro query. Por ejemplo, <BIGQUERY_PROJECT>:<DATASET_NAME>.<INPUT_TABLE>.
outputDeadletterTable la tabla de BigQuery de los mensajes que no se han podido enviar a la tabla de salida. 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 de SQL que se usará para leer datos de BigQuery. Si el conjunto de datos de BigQuery está en un proyecto diferente al del trabajo de Dataflow, especifica el nombre completo del conjunto de datos en la consulta 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 defines ambos parámetros, la plantilla usará el parámetro query. Por ejemplo, select * from sampledb.sample_table.
useLegacySql asigna el valor true para usar SQL antiguo. Este parámetro solo se aplica cuando se usa el parámetro query. El valor predeterminado es false.
queryLocation se necesita al leer datos de una vista autorizada sin el permiso de la tabla subyacente. Por ejemplo, US.
queryTempDataset con esta opción, puedes definir un conjunto de datos para crear la tabla temporal en la que se almacenarán los resultados de la consulta. Por ejemplo, temp_dataset.
KMSEncryptionKey si lees datos de BigQuery mediante una fuente de consulta, usa esta clave de Cloud KMS para cifrar las tablas temporales que se creen. Por ejemplo, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
elasticsearchUsername: nombre de usuario de Elasticsearch para autenticarte. Si se especifica, se ignora el valor de apiKey.
elasticsearchPassword la contraseña de Elasticsearch para autenticarse. Si se especifica, se ignora el valor de apiKey.
batchSize el tamaño del lote en número de documentos. El valor predeterminado es 1000.
batchSizeBytes tamaño del lote en número de bytes. El valor predeterminado es 5242880 (5 MB).
maxRetryAttempts número máximo de reintentos. Debe ser mayor que cero. El valor predeterminado es no retries.
maxRetryDuration la duración máxima de los reintentos en milisegundos. Debe ser mayor que cero. El valor predeterminado es no retries.
propertyAsIndex la propiedad del documento que se está indexando cuyo valor especifica los metadatos de _index que se incluirán con el documento en las solicitudes masivas. Tiene prioridad sobre una función definida por el usuario _index. El valor predeterminado es none.
javaScriptIndexFnGcsPath la ruta de Cloud Storage al origen de la función UDF de JavaScript que especifica los metadatos de _index que se incluirán con el documento en las solicitudes masivas. El valor predeterminado es none.
javaScriptIndexFnName nombre de la función JavaScript de UDF que especifica los metadatos de _index que se incluirán con el documento en las solicitudes masivas. El valor predeterminado es none.
propertyAsId una propiedad del documento que se está indexando cuyo valor especifica los metadatos de _id que se incluirán con el documento en las solicitudes masivas. Tiene prioridad sobre una función definida por el usuario _id. El valor predeterminado es none.
javaScriptIdFnGcsPath la ruta de Cloud Storage a la fuente de la función UDF de JavaScript que especifica los metadatos de _id que se incluirán con el documento en las solicitudes masivas. El valor predeterminado es none.
javaScriptIdFnName nombre de la función JavaScript de UDF que especifica los metadatos _id que se incluirán con el documento en las solicitudes masivas. El valor predeterminado es none.
javaScriptTypeFnGcsPath la ruta de Cloud Storage al origen de la función UDF de JavaScript de una función que especifica los metadatos de _type que se deben incluir con los documentos en las solicitudes masivas. El valor predeterminado es none.
javaScriptTypeFnName nombre de la función JavaScript de la UDF que especifica los metadatos de _type que se incluirán con el documento en las solicitudes masivas. El valor predeterminado es none.
javaScriptIsDeleteFnGcsPath la ruta de Cloud Storage al origen de la función UDF de JavaScript que determina si se debe eliminar el documento en lugar de insertarlo o actualizarlo. La función devuelve un valor de cadena de true o false. El valor predeterminado es none.
javaScriptIsDeleteFnName el nombre de la función JavaScript de la FDU que determina si se debe eliminar el documento en lugar de insertarlo o actualizarlo. La función devuelve un valor de cadena de true o false. El valor predeterminado es none.
usePartialUpdate indica si se deben usar actualizaciones parciales (actualizar en lugar de crear o indexar, lo que permite documentos parciales) con las solicitudes de Elasticsearch. El valor predeterminado es false.
bulkInsertMethod indica si se debe usar INDEX (índice, permite upserts) o CREATE (crear, errores en _id duplicados) con solicitudes masivas de Elasticsearch. El valor predeterminado es CREATE.
trustSelfSignedCerts indica si se debe confiar en los certificados autofirmados. Una instancia de Elasticsearch instalada puede tener un certificado autofirmado. Habilita esta opción para omitir la validación del 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 autofirmado. Para omitir la validación del certificado, asigna el valor true a este parámetro. El valor predeterminado es false.
apiKeyKMSEncryptionKey la clave de Cloud KMS para descifrar la clave de API. Este parámetro es obligatorio si apiKeySource tiene el valor KMS. Si se proporciona este parámetro, debe incluir una cadena apiKey cifrada. Encripta los parámetros con el endpoint de encriptación de la API 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 de secreto de Secret Manager de la apiKey. Si apiKeySource tiene el valor SECRET_MANAGER, proporcione 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 la fuente de la clave de API. Los valores permitidos son PLAINTEXT, KMS y SECRET_MANAGER. Este parámetro es obligatorio cuando se usa Secret Manager o KMS. Si apiKeySource se define como KMS, se deben proporcionar apiKeyKMSEncryptionKey y la clave de API cifrada. Si apiKeySource se define como SECRET_MANAGER, se debe proporcionar apiKeySecretId. Si apiKeySource se define como PLAINTEXT, se debe proporcionar apiKey. El valor predeterminado es PLAINTEXT.
socketTimeout si se define, sobrescribe el tiempo de espera máximo de reintento predeterminado y el tiempo de espera de socket predeterminado (30.000 ms) en Elastic RestClient.
javascriptTextTransformGcsPath el URI de Cloud Storage del archivo .js que define la función de JavaScript definida por el usuario (UDF) que se va a usar. Por ejemplo, gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName nombre de la función definida por el usuario (UDF) de JavaScript que se va a usar. Por ejemplo, si el código de la función de JavaScript es myTransform(inJson) { /*...do stuff...*/ }, el nombre de la función es myTransform. Para ver ejemplos de UDFs de JavaScript, consulta Ejemplos de UDFs (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples).

Funciones definidas por el usuario

Esta plantilla admite funciones definidas por el usuario (UDF) en varios puntos del flujo de procesamiento, que se describen a continuación. Para obtener más información, consulta el artículo sobre cómo crear funciones definidas por el usuario para plantillas de Dataflow.

Función de índice

Devuelve el índice al que pertenece el documento.

Parámetros de plantilla:

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

Especificación de la función:

Entrada: el documento de Elasticsearch, serializado como una cadena JSON.
Salida: el valor del campo de metadatos _index del documento.

Función de ID de documento

Devuelve el ID del documento.

Parámetros de plantilla:

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

Especificación de la función:

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

Función de eliminación de documentos

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

Parámetros de plantilla:

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

Especificación de la función:

Entrada: el documento de Elasticsearch, serializado como una cadena JSON.
Salida: devuelve la cadena "true" para eliminar el documento o "false" para insertarlo o actualizarlo.

Función de tipo de asignación

Devuelve el tipo de asignación del documento.

Parámetros de plantilla:

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

Especificación de la función:

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

Ejecutar la plantilla

Consola

Ve a la página Crear tarea a partir de plantilla de Dataflow.

Ir a Crear tarea a partir de plantilla

En el campo Nombre de la tarea, introduce un nombre único.
Opcional: En Endpoint regional, seleccione un valor en el menú desplegable. La región predeterminada es us-central1.
Para ver una lista de las regiones en las que puedes ejecutar una tarea de Dataflow, consulta Ubicaciones de Dataflow.
En el menú desplegable Plantilla de flujo de datos, seleccione the BigQuery to Elasticsearch template.
En los campos de parámetros proporcionados, introduzca los valores de los parámetros.
Haz 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/BigQuery_to_Elasticsearch \
    --parameters \
inputTableSpec=INPUT_TABLE_SPEC,\
connectionUrl=CONNECTION_URL,\
apiKey=APIKEY,\
index=INDEX

Haz los cambios siguientes:

PROJECT_ID: el ID del proyecto Google Cloud en el que quieres ejecutar la tarea de Dataflow
JOB_NAME: un nombre de trabajo único que elijas
REGION_NAME: la región en la que quieras desplegar tu trabajo de Dataflow. Por ejemplo, us-central1
VERSION: la versión de la plantilla que quieres usar
Puedes usar los siguientes valores:
- latest para usar la última versión de la plantilla, que está disponible en la carpeta principal sin fecha del contenedor: 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 encuentra anidada en la carpeta principal correspondiente con la fecha en el bucket: gs://dataflow-templates-REGION_NAME/
Precaución: La última versión de las plantillas puede actualizarse con cambios importantes. Tus entornos de producción deben usar plantillas que se encuentren en la carpeta principal con fecha más reciente para evitar que estos cambios afecten a tus flujos de trabajo de producción.
INPUT_TABLE_SPEC: el nombre de la tabla de BigQuery.
CONNECTION_URL: la URL de Elasticsearch.
APIKEY: tu clave de API codificada en base64 para la autenticación.
INDEX: tu índice de Elasticsearch.

API

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

Haz los cambios siguientes:

PROJECT_ID: el ID del proyecto Google Cloud en el que quieres ejecutar la tarea de Dataflow
JOB_NAME: un nombre de trabajo único que elijas
LOCATION: la región en la que quieras desplegar tu trabajo de Dataflow. Por ejemplo, us-central1
VERSION: la versión de la plantilla que quieres usar
Puedes usar los siguientes valores:
- latest para usar la última versión de la plantilla, que está disponible en la carpeta principal sin fecha del contenedor: 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 encuentra anidada en la carpeta principal correspondiente con la fecha en el bucket: gs://dataflow-templates-REGION_NAME/
Precaución: La última versión de las plantillas puede actualizarse con cambios importantes. Tus entornos de producción deben usar plantillas que se encuentren en la carpeta principal con fecha más reciente para evitar que estos cambios afecten a tus flujos de trabajo de producción.
INPUT_TABLE_SPEC: el nombre de la tabla de BigQuery.
CONNECTION_URL: la URL de Elasticsearch.
APIKEY: tu clave de API codificada en base64 para la autenticación.
INDEX: tu í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();
  }
}

Siguientes pasos

Consulta información sobre las plantillas de Dataflow.
Consulta la lista de plantillas proporcionadas por Google.

Plantilla de BigQuery a Elasticsearch Organízate con las colecciones Guarda y clasifica el contenido según tus preferencias.

Requisitos del flujo de procesamiento

Parámetros de plantilla

Parámetros obligatorios

Parámetros opcionales

Funciones definidas por el usuario

Función de índice

Función de ID de documento

Función de eliminación de documentos

Función de tipo de asignación

Ejecutar la plantilla

Consola

gcloud

API

Código fuente de la plantilla

Java

Siguientes pasos

Plantilla de BigQuery a Elasticsearch