Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

Modello BigQuery a Elasticsearch

Il modello da BigQuery a Elasticsearch è una pipeline batch che importa dati da una tabella BigQuery in Elasticsearch come documenti. Il modello può leggere l'intera tabella o record specifici utilizzando una query fornita.

Requisiti della pipeline

La tabella BigQuery di origine deve esistere.
Un host Elasticsearch su un' Google Cloud istanza o su Elastic Cloud con Elasticsearch versione 7.0 o successive. Deve essere accessibile dalle macchine worker Dataflow.

Parametri del modello

Parametri obbligatori

connectionUrl: l'URL di Elasticsearch nel formato https://hostname:[port]. Se utilizzi Elastic Cloud, specifica CloudID. Ad esempio, https://elasticsearch-host:9200.
apiKey: la chiave API codificata in Base64 da utilizzare per l'autenticazione.
index: l'indice Elasticsearch a cui vengono inviate le richieste. Ad esempio, my-index.

Parametri facoltativi

inputTableSpec: la tabella BigQuery da cui leggere. Se specifichi inputTableSpec, il modello legge i dati direttamente dallo spazio di archiviazione BigQuery utilizzando l'API BigQuery Storage Read (https://cloud.google.com/bigquery/docs/reference/storage). Per informazioni sulle limitazioni dell'API Storage Read, consulta https://cloud.google.com/bigquery/docs/reference/storage#limitations. Devi specificare inputTableSpec o query. Se imposti entrambi i parametri, il modello utilizza il parametro query. Ad esempio, <BIGQUERY_PROJECT>:<DATASET_NAME>.<INPUT_TABLE>.
outputDeadletterTable: la tabella BigQuery per i messaggi che non sono riusciti a raggiungere la tabella di output. Se una tabella non esiste, viene creata durante l'esecuzione della pipeline. Se non viene specificato, viene utilizzato <outputTableSpec>_error_records. Ad esempio, <PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE>.
query: la query SQL da utilizzare per leggere i dati da BigQuery. Se il set di dati BigQuery si trova in un progetto diverso dal job Dataflow, specifica il nome completo del set di dati nella query SQL, ad esempio: <PROJECT_ID>.<DATASET_NAME>.<TABLE_NAME>. Per impostazione predefinita, il parametro query utilizza GoogleSQL (https://cloud.google.com/bigquery/docs/introduction-sql), a meno che useLegacySql non sia true. Devi specificare inputTableSpec o query. Se imposti entrambi i parametri, il modello utilizza il parametro query. Ad esempio, select * from sampledb.sample_table.
useLegacySql: imposta su true per utilizzare SQL precedente. Questo parametro si applica solo quando si utilizza il parametro query. Il valore predefinito è false.
queryLocation: necessario quando si legge da una vista autorizzata senza l'autorizzazione della tabella sottostante. Ad esempio, US.
queryTempDataset: con questa opzione, puoi impostare un set di dati esistente per creare la tabella temporanea in cui archiviare i risultati della query. Ad esempio, temp_dataset.
KMSEncryptionKey: se leggi da BigQuery utilizzando l'origine della query, utilizza questa chiave Cloud KMS per criptare le tabelle temporanee create. Ad esempio, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
elasticsearchUsername: il nome utente di Elasticsearch per l'autenticazione. Se specificato, il valore di apiKey viene ignorato.
elasticsearchPassword: la password di Elasticsearch per l'autenticazione. Se specificato, il valore di apiKey viene ignorato.
batchSize: le dimensioni del batch in numero di documenti. Il valore predefinito è 1000.
batchSizeBytes: le dimensioni del batch in numero di byte. Il valore predefinito è 5242880 (5 MB).
maxRetryAttempts: il numero massimo di tentativi. Deve essere maggiore di zero. Il valore predefinito è 5.
maxRetryDuration: la durata massima dei tentativi in millisecondi. Deve essere maggiore di zero. Il valore predefinito è 60000 (1 minuto).
propertyAsIndex: la proprietà nel documento indicizzato il cui valore specifica i metadati _index da includere nel documento nelle richieste collettive. Ha la precedenza su una funzione definita dall'utente _index. Il valore predefinito è none.
javaScriptIndexFnGcsPath: il percorso Cloud Storage all'origine della funzione JavaScript definita dall'utente per una funzione che specifica i metadati _index da includere nel documento nelle richieste collettive. Il valore predefinito è none.
javaScriptIndexFnName: il nome della funzione JavaScript definita dall'utente che specifica i metadati _index da includere nel documento nelle richieste collettive. Il valore predefinito è none.
propertyAsId: una proprietà nel documento indicizzato il cui valore specifica i metadati _id da includere nel documento nelle richieste collettive. Ha la precedenza su una funzione definita dall'utente _id. Il valore predefinito è none.
javaScriptIdFnGcsPath: il percorso Cloud Storage all'origine della funzione JavaScript definita dall'utente per la funzione che specifica i metadati _id da includere nel documento nelle richieste collettive. Il valore predefinito è none.
javaScriptIdFnName: il nome della funzione JavaScript definita dall'utente che specifica i metadati _id da includere nel documento nelle richieste collettive. Il valore predefinito è none.
javaScriptTypeFnGcsPath: il percorso Cloud Storage all'origine della funzione JavaScript definita dall'utente per una funzione che specifica i metadati _type da includere nei documenti nelle richieste collettive. Il valore predefinito è none.
javaScriptTypeFnName: il nome della funzione JavaScript definita dall'utente che specifica i metadati _type da includere nel documento nelle richieste collettive. Il valore predefinito è none.
javaScriptIsDeleteFnGcsPath: il percorso Cloud Storage all'origine della funzione JavaScript definita dall'utente per la funzione che determina se eliminare il documento anziché inserirlo o aggiornarlo. La funzione restituisce un valore stringa true o false. Il valore predefinito è none.
javaScriptIsDeleteFnName: il nome della funzione JavaScript definita dall'utente che determina se eliminare il documento anziché inserirlo o aggiornarlo. La funzione restituisce un valore stringa true o false. Il valore predefinito è none.
usePartialUpdate: indica se utilizzare gli aggiornamenti parziali (aggiorna anziché creare o indicizzare, consentendo documenti parziali) con le richieste Elasticsearch. Il valore predefinito è false.
bulkInsertMethod: indica se utilizzare INDEX (indice, consente upsert) o CREATE (crea, errori su _id duplicato) con le richieste collettive Elasticsearch. Il valore predefinito è CREATE.
trustSelfSignedCerts: indica se considerare attendibile o meno il certificato autofirmato. Un'istanza Elasticsearch installata potrebbe avere un certificato autofirmato. Imposta questo valore su true per ignorare la convalida del certificato SSL. (Il valore predefinito è false).
disableCertificateValidation: se true, considera attendibile il certificato SSL autofirmato. Un'istanza Elasticsearch potrebbe avere un certificato autofirmato. Per ignorare la convalida del certificato, imposta questo parametro su true. Il valore predefinito è false.
apiKeyKMSEncryptionKey: la chiave Cloud KMS per decriptare la chiave API. Questo parametro è obbligatorio se apiKeySource è impostato su KMS. Se questo parametro viene fornito, inserisci una stringa apiKey criptata. Cripta i parametri utilizzando l'endpoint di criptaggio dell'API KMS. Per la chiave, utilizza il 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 Ad esempio, projects/your-project-id/locations/global/keyRings/your-keyring/cryptoKeys/your-key-name.
apiKeySecretId: l'ID secret di Secret Manager per apiKey. Se apiKeySource è impostato su SECRET_MANAGER, fornisci questo parametro. Utilizza il formato projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. For example, projects/your-project-id/secrets/your-secret/versions/your-secret-version`.
apiKeySource: l'origine della chiave API. I valori consentiti sono PLAINTEXT, KMS oand SECRET_MANAGER. Questo parametro è obbligatorio quando utilizzi Secret Manager o KMS. Se apiKeySource è impostato su KMS, devono essere forniti apiKeyKMSEncryptionKey e apiKey criptata. Se apiKeySource è impostato su SECRET_MANAGER, deve essere fornito apiKeySecretId. Se apiKeySource è impostato su PLAINTEXT, deve essere fornito apiKey. Il valore predefinito è PLAINTEXT.
socketTimeout: se impostato, sovrascrive il timeout massimo predefinito per i tentativi e il timeout predefinito del socket (30000 ms) in Elastic RestClient.
javascriptTextTransformGcsPath: l'URI Cloud Storage del file .js che definisce la funzione JavaScript definita dall'utente (UDF) da utilizzare. Ad esempio, gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName: il nome della funzione definita dall'utente (UDF) JavaScript da utilizzare. Ad esempio, se il codice della funzione JavaScript è myTransform(inJson) { /*...do stuff...*/ }, il nome della funzione è myTransform. Per esempi di funzioni JavaScript definite dall'utente, consulta Esempi di funzioni definite dall'utente (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples).

Funzioni definite dall'utente

Questo modello supporta le funzioni definite dall'utente (UDF) in diversi punti della pipeline, descritti di seguito. Per ulteriori informazioni, consulta Creare funzioni definite dall'utente per i modelli Dataflow.

Funzione indice

Restituisce l'indice a cui appartiene il documento.

Parametri del modello:

javaScriptIndexFnGcsPath: l'URI Cloud Storage del file JavaScript.
javaScriptIndexFnName: il nome della funzione JavaScript.

Specifica della funzione:

Input: il documento Elasticsearch, serializzato come stringa JSON.
Output: il valore del campo dei metadati _index del documento.

Funzione ID documento

Restituisce l'ID documento.

Parametri del modello:

javaScriptIdFnGcsPath: l'URI Cloud Storage del file JavaScript.
javaScriptIdFnName: il nome della funzione JavaScript.

Specifica della funzione:

Input: il documento Elasticsearch, serializzato come stringa JSON.
Output: il valore del campo dei metadati _id del documento.

Funzione di eliminazione dei documenti

Specifica se eliminare un documento. Per utilizzare questa funzione, imposta la modalità di inserimento collettivo su INDEX e fornisci una funzione ID documento.

Parametri del modello:

javaScriptIsDeleteFnGcsPath: l'URI Cloud Storage del file JavaScript.
javaScriptIsDeleteFnName: il nome della funzione JavaScript.

Specifica della funzione:

Input: il documento Elasticsearch, serializzato come stringa JSON.
Output: restituisce la stringa "true" per eliminare il documento o "false" per eseguire l'upsert del documento.

Funzione tipo di mappatura

Restituisce il tipo di mappatura del documento.

Parametri del modello:

javaScriptTypeFnGcsPath: l'URI Cloud Storage del file JavaScript.
javaScriptTypeFnName: il nome della funzione JavaScript.

Specifica della funzione:

Input: il documento Elasticsearch, serializzato come stringa JSON.
Output: il valore del campo dei metadati _type del documento.

Esegui il modello

Console

Vai alla pagina Crea job da modello di Dataflow.

Vai a Crea job da modello

Nel campo Nome job, inserisci un nome job univoco.
(Facoltativo) Per Endpoint regionale, seleziona un valore dal menu a discesa. La regione predefinita è us-central1.
Per un elenco delle regioni in cui puoi eseguire un job Dataflow, consulta Località di Dataflow.
Dal menu a discesa Modello Dataflow, seleziona il modello Da BigQuery a Elasticsearch.
Nei campi dei parametri forniti, inserisci i valori dei parametri.
Fai clic su Esegui job.

gcloud

Nella shell o nel terminale, esegui il modello:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/ \
    --parameters \
inputTableSpec=INPUT_TABLE_SPEC,\
connectionUrl=CONNECTION_URL,\
apiKey=APIKEY,\
index=INDEX

Sostituisci quanto segue:

PROJECT_ID: l'ID progetto in cui vuoi eseguire il job Dataflow Google Cloud
JOB_NAME: un nome job univoco a tua scelta
REGION_NAME: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
VERSION: la versione del modello che vuoi utilizzare
Puoi utilizzare i seguenti valori:
- latest per utilizzare la versione più recente del modello, disponibile nella cartella principale non datata nel bucket: gs://dataflow-templates-REGION_NAME/latest/
- il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella principale datata nel bucket: gs://dataflow-templates-REGION_NAME/
Attenzione: La versione più recente dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale datata più recente per evitare che queste modifiche che causano interruzioni influiscano sui flussi di lavoro di produzione.
INPUT_TABLE_SPEC: il nome della tabella BigQuery.
CONNECTION_URL: l'URL di Elasticsearch.
APIKEY: la chiave API codificata in Base64 per l'autenticazione.
INDEX: l'indice Elasticsearch.

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta HTTP POST. Per ulteriori informazioni sull' API e sui relativi ambiti di autorizzazione, 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/",
   }
}

Sostituisci quanto segue:

PROJECT_ID: l'ID progetto in cui vuoi eseguire il job Dataflow Google Cloud
JOB_NAME: un nome job univoco a tua scelta
LOCATION: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
VERSION: la versione del modello che vuoi utilizzare
Puoi utilizzare i seguenti valori:
- latest per utilizzare la versione più recente del modello, disponibile nella cartella principale non datata nel bucket: gs://dataflow-templates-REGION_NAME/latest/
- il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella principale datata nel bucket: gs://dataflow-templates-REGION_NAME/
Attenzione: La versione più recente dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale datata più recente per evitare che queste modifiche che causano interruzioni influiscano sui flussi di lavoro di produzione.
INPUT_TABLE_SPEC: il nome della tabella BigQuery.
CONNECTION_URL: l'URL di Elasticsearch.
APIKEY: la chiave API codificata in Base64 per l'autenticazione.
INDEX: l'indice Elasticsearch.

Codice sorgente del modello

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

Passaggi successivi

Scopri di più sui modelli Dataflow.
Consulta l'elenco dei modelli forniti da Google.

Modello BigQuery a Elasticsearch Mantieni tutto organizzato con le raccolte Salva e classifica i contenuti in base alle tue preferenze.

Requisiti della pipeline

Parametri del modello

Parametri obbligatori

Parametri facoltativi

Funzioni definite dall'utente

Funzione indice

Funzione ID documento

Funzione di eliminazione dei documenti

Funzione tipo di mappatura

Esegui il modello

Console

gcloud

API

Codice sorgente del modello

Java

Passaggi successivi

Modello BigQuery a Elasticsearch