Modelo do BigQuery para Elasticsearch

O modelo do BigQuery para Elasticsearch é um pipeline em lote que ingere dados de uma tabela do BigQuery no Elasticsearch como documentos. O modelo pode ler toda a tabela ou ler registros específicos usando uma consulta fornecida.

Requisitos de pipeline

A tabela de origem do BigQuery precisa existir.
Um host do Elasticsearch em uma instância do Google Cloud ou no Elastic Cloud com o Elasticsearch versão 7.0 ou mais recente. Precisa ser acessível nas máquinas de worker do Dataflow.

Parâmetros do modelo

Parâmetro	Descrição
`connectionUrl`	URL do Elasticsearch no formato `https://hostname:[port]` ou especifique o CloudID se estiver usando o Elastic Cloud.
`apiKey`	Chave da API codificada em Base64 usada para autenticação.
`index`	O índice do Elasticsearch para onde as solicitações serão emitidas, por exemplo, `my-index`.
`inputTableSpec`	(Opcional) Tabela do BigQuery de onde ler para inserir no Elasticsearch. Forneça a tabela ou a consulta. Por exemplo, `projectId:datasetId.tablename`
`query`	(Opcional) Consulta SQL para extrair dados do BigQuery. Forneça a tabela ou a consulta.
`useLegacySql`	(Opcional) Defina como "true" para usar o SQL legado (somente aplicável se a consulta for fornecida). Padrão: `false`.
`batchSize`	(Opcional) Tamanho do lote em número de documentos. Padrão: `1000`.
`batchSizeBytes`	(Opcional) Tamanho do lote em número de bytes. Padrão: `5242880` (5 mb).
`maxRetryAttempts`	(Opcional) Máximo de tentativas de repetição. Precisa ser > 0. Padrão: `no retries`.
`maxRetryDuration`	(Opcional) A duração máxima da nova tentativa em milissegundos precisa ser maior que 0. Padrão: `no retries`.
`propertyAsIndex`	(Opcional) Uma propriedade no documento que está sendo indexado com o valor que especificará os metadados de `_index` a serem incluídos com o documento na solicitação em massa (tem precedência sobre uma UDF `_index`). Padrão: none.
`propertyAsId`	(Opcional) Uma propriedade no documento que está sendo indexado com o valor que especificará os metadados de `_id` a serem incluídos com o documento na solicitação em massa (tem precedência sobre uma UDF `_id`). Padrão: none.
`javaScriptIndexFnGcsPath`	(Opcional) O caminho do Cloud Storage para a origem UDF em JavaScript de uma função que especificará os metadados de `_index` a serem incluídos com o documento na solicitação em massa. Padrão: none.
`javaScriptIndexFnName`	(Opcional) Nome da função UDF em JavaScript para a função que especificará os metadados de `_index` a serem incluídos com o documento na solicitação em massa. Padrão: none.
`javaScriptIdFnGcsPath`	(Opcional) O caminho do Cloud Storage para a origem UDF em JavaScript de uma função que especificará os metadados de `_id` a serem incluídos com o documento na solicitação em massa. Padrão: none.
`javaScriptIdFnName`	(Opcional) Nome da função UDF em JavaScript para a função que especificará os metadados de `_id` a serem incluídos com o documento na solicitação em massa. Padrão: none.
`javaScriptTypeFnGcsPath`	(Opcional) O caminho do Cloud Storage para a origem UDF em JavaScript de uma função que especificará os metadados de `_type` a serem incluídos com o documento na solicitação em massa. Padrão: none.
`javaScriptTypeFnName`	(Opcional) Nome da função UDF em JavaScript para a função que especificará os metadados de `_type` a serem incluídos com o documento na solicitação em massa. Padrão: none.
`javaScriptIsDeleteFnGcsPath`	(Opcional) O caminho do Cloud Storage para a origem UDF em JavaScript de uma função que determina se o documento deve ser excluído em vez de inserido ou atualizado. A função precisa retornar o valor de string `"true"` ou `"false"`. Padrão: none.
`javaScriptIsDeleteFnName`	(Opcional) Nome da função UDF em JavaScript de uma função que vai determinar se o documento deve ser excluído em vez de inserido ou atualizado. A função precisa retornar o valor de string `"true"` ou `"false"`. Padrão: none.
`usePartialUpdate`	(Opcional) Indica se as atualizações parciais vão ser usadas (atualizar em vez de criar ou indexar, permitindo documentos parciais) com solicitações Elasticsearch. Padrão: `false`.
`bulkInsertMethod`	(Opcional) Indica se é necessário usar `INDEX` (índice, permite ajustes) ou `CREATE` (criar, erros em _id duplicados) com solicitações em massa do Elasticsearch. Padrão: `CREATE`.

Funções definidas pelo usuário

Esse modelo é compatível com funções definidas pelo usuário (UDFs) em vários pontos do pipeline, descritas abaixo. Para mais informações, consulte Criar funções definidas pelo usuário para modelos do Dataflow.

Função de índice

Retorna o índice ao qual o documento pertence.

Parâmetros do modelo:

javaScriptIndexFnGcsPath: o URI do Cloud Storage do arquivo JavaScript.
javaScriptIndexFnName: o nome da função JavaScript.

Especificação da função:

Entrada: o documento do Elasticsearch, serializado como uma string JSON.
Saída: o valor do campo de metadados _index do documento.

Função ID do documento

Retorna o ID do documento.

Parâmetros do modelo:

javaScriptIdFnGcsPath: o URI do Cloud Storage do arquivo JavaScript.
javaScriptIdFnName: o nome da função JavaScript.

Especificação da função:

Entrada: o documento do Elasticsearch, serializado como uma string JSON.
Saída: o valor do campo de metadados _id do documento.

Função de exclusão de documentos

Especifica se um documento deve ser excluído. Para usar essa função, defina o modo de inserção em massa como INDEX e forneça uma função de ID do documento.

Parâmetros do modelo:

javaScriptIsDeleteFnGcsPath: o URI do Cloud Storage do arquivo JavaScript.
javaScriptIsDeleteFnName: o nome da função JavaScript.

Especificação da função:

Entrada: o documento do Elasticsearch, serializado como uma string JSON.
Saída: retorna a string "true" para excluir o documento ou "false" para manter o documento.

Função do tipo de mapeamento

Retorna o tipo de mapeamento do documento.

Parâmetros do modelo:

javaScriptTypeFnGcsPath: o URI do Cloud Storage do arquivo JavaScript.
javaScriptTypeFnName: o nome da função JavaScript.

Especificação da função:

Entrada: o documento do Elasticsearch, serializado como uma string JSON.
Saída: o valor do campo de metadados _type do documento.

Executar o modelo

Console

Acesse a página Criar job usando um modelo do Dataflow.

Acesse Criar job usando um modelo

No campo Nome do job, insira um nome exclusivo.
Opcional: em Endpoint regional, selecione um valor no menu suspenso. A região padrão é us-central1.
Para ver uma lista de regiões em que é possível executar um job do Dataflow, consulte Locais do Dataflow.
No menu suspenso Modelo do Dataflow, selecione the BigQuery to Elasticsearch template.
Nos campos de parâmetro fornecidos, insira os valores de parâmetro.
Cliquem em Executar job.

gcloud

No shell ou no terminal, execute o modelo:

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

Substitua:

PROJECT_ID: o ID do projeto do Google Cloud em que você quer executar o job do Dataflow
JOB_NAME: um nome de job de sua escolha
REGION_NAME: a região onde você quer implantar o job do Dataflow, por exemplo, us-central1
VERSION: a versão do modelo que você quer usar
Use estes valores:
- latest para usar a versão mais recente do modelo, disponível na pasta mãe não datada no bucket: gs://dataflow-templates-REGION_NAME/latest/
- o nome da versão, como 2023-09-12-00_RC00, para usar uma versão específica do modelo, que pode ser aninhada na respectiva pasta mãe datada no bucket: gs://dataflow-templates-REGION_NAME/
Cuidado: é possível que a versão mais recente dos modelos seja atualizada com alterações interruptivas. Os ambientes de produção precisam usar os modelos mantidos na pasta mãe datada mais recente para evitar que essas alterações interruptivas afetem os fluxos de trabalho de produção.
INPUT_TABLE_SPEC: o nome da tabela do BigQuery
CONNECTION_URL: seu URL do Elasticsearch
APIKEY: sua chave de API codificada em base64 para autenticação.
INDEX: seu índice do Elasticsearch.

API

Para executar o modelo usando a API REST, envie uma solicitação HTTP POST. Para mais informações sobre a API e os respectivos escopos de autorização, consulte 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",
   }
}

Substitua:

PROJECT_ID: o ID do projeto do Google Cloud em que você quer executar o job do Dataflow
JOB_NAME: um nome de job de sua escolha
LOCATION: a região onde você quer implantar o job do Dataflow, por exemplo, us-central1
VERSION: a versão do modelo que você quer usar
Use estes valores:
- latest para usar a versão mais recente do modelo, disponível na pasta mãe não datada no bucket: gs://dataflow-templates-REGION_NAME/latest/
- o nome da versão, como 2023-09-12-00_RC00, para usar uma versão específica do modelo, que pode ser aninhada na respectiva pasta mãe datada no bucket: gs://dataflow-templates-REGION_NAME/
Cuidado: é possível que a versão mais recente dos modelos seja atualizada com alterações interruptivas. Os ambientes de produção precisam usar os modelos mantidos na pasta mãe datada mais recente para evitar que essas alterações interruptivas afetem os fluxos de trabalho de produção.
INPUT_TABLE_SPEC: o nome da tabela do BigQuery
CONNECTION_URL: seu URL do Elasticsearch
APIKEY: sua chave de API codificada em base64 para autenticação.
INDEX: seu índice do Elasticsearch.

Código-fonte do modelo

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

A seguir

Saiba mais sobre os modelos do Dataflow.
Confira a lista de modelos fornecidos pelo Google.