Vorlage „BigQuery für Elasticsearch“

Beta

Diese Funktion unterliegt den „Nutzungsbedingungen für Pre-GA-Angebote“ im Abschnitt „Allgemeine Nutzungsbedingungen“ der dienstspezifischen Nutzungsbedingungen. Pre-GA-Funktionen sind „wie besehen“ verfügbar und werden möglicherweise nur eingeschränkt unterstützt. Weitere Informationen finden Sie in den Beschreibungen der Startphase.

Die Vorlage "BigQuery für Elasticsearch" ist eine Batchpipeline, die Daten aus einer BigQuery-Tabelle als Dokumente in Elasticsearch aufnimmt. Die Vorlage kann entweder die gesamte Tabelle oder bestimmte Datensätze mithilfe einer angegebenen Abfrage lesen.

Pipelineanforderungen

• Die BigQuery-Quelltabelle
• Ein Elasticsearch-Host auf einer Google Cloud-Instanz oder in Elastic Cloud mit Elasticsearch Version 7.0 oder höher. Muss über die Dataflow-Worker-Maschinen zugänglich sein.

Vorlagenparameter

Erforderliche Parameter

• connectionUrl : Elasticsearch-URL im Format https://hostname:[port] oder bei Verwendung von Elastic Cloud (Beispiel: https://elasticsearch-host:9200) eine CloudID angeben.
• apiKey : Base64-codierter API-Schlüssel für die Authentifizierung. Informationen hierzu unter: https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html#security-api-create-api-key-request “
• index : Index, an den die Anfragen gestellt werden (Beispiel: my-index)

Optionale Parameter

• inputTableSpec : Spezifikation der BigQuery-Quelltabelle. (Beispiel: bigquery-project:dataset.input_table).
• outputDeadletterTable : Nachrichten, die die Ausgabetabelle aus irgendeinem Grund nicht erreicht haben (z.B. nicht übereinstimmendes Schema, fehlerhaft formatierte JSON-Datei), werden in diese Tabelle geschrieben. Wenn sie nicht vorhanden ist, wird sie während der Pipelineausführung erstellt. (Beispiel: Ihre-Projekt-ID:Ihr-Dataset.Ihr-Tabellenname).
• Abfrage : Abfrage, die in der Quelle zur Extraktion der Daten ausgeführt. (Beispiel: Wählen Sie "*" aus sampledb.sample_table aus).
• useLegacySql : Legen Sie den Wert "true" fest, um Legacy-SQL zu verwenden (nur anwendbar, wenn eine Abfrage bereitstellt wird). Die Standardeinstellung ist "false".
• queryLocation : Erforderlich, wenn aus einer autorisierten Ansicht ohne die Berechtigung der zugrunde liegenden Tabelle gelesen wird. (Beispiel: US).
• elasticsearchUsername : Der Elasticsearch-Nutzername für die Authentifizierung. Wenn angegeben, wird der Wert von "apiKey" ignoriert.
• elasticsearchPassword : Das Elasticsearch-Passwort, mit dem die Authentifizierung erfolgen soll. Wenn angegeben, wird der Wert von "apiKey" ignoriert.
• batchSize : Batchgröße in der Anzahl an Dokumenten. Standardeinstellung: '1000'.
• batchSizeBytes : Batchgröße in Byte für die Batch-Einfügung von Nachrichten in Elasticsearch. Standardeinstellung: '5242880 (5mb)'.
• maxRetryAttempts : Maximale Wiederholungsversuche, muss > 0 sein. Standardeinstellung: 'no retries'.
• maxRetryDuration : Maximale Wiederholungsdauer in Millisekunden, muss > 0 sein. Standardeinstellung: 'no retries'.
• propertyAsIndex : Ein Attribut im indexierten Dokument, dessen Wert angibt, dass "_index"-Metadaten im Dokument in der Bulk-Anfrage enthalten sein sollen (hat Vorrang vor einer "_index"-UDF). Standardwert: none.
• javaScriptIndexFnGcsPath : Der Cloud Storage-Pfad zur JavaScript-UDF-Quelle für eine Funktion, die angibt, dass „_index“-Metadaten im Dokument in der Bulk-Anfrage enthalten sein sollen.Standard: none.
• javaScriptIndexFnName : UDF-JavaScript-Funktionsname für eine Funktion, die angibt, dass _index-Metadaten im Dokument in der Bulk-Anfrage enthalten sein sollen. Standardwert: none.
• propertyAsId : Ein Attribut im indexierten Dokument, dessen Wert angibt, dass "_id"-Metadaten im Dokument in der Bulk-Anfrage enthalten sein sollen (hat Vorrang vor einer "_id"-UDF). Standardwert: none.
• javaScriptIdFnGcsPath : Der Cloud Storage-Pfad zur JavaScript-UDF-Quelle für eine Funktion, die angibt, dass „_id“-Metadaten im Dokument in der Bulk-Anfrage enthalten sein sollen.Standard: none.
• javaScriptIdFnName : UDF-JavaScript-Funktionsname für eine Funktion, die angibt, dass _id-Metadaten im Dokument in der Bulk-Anfrage enthalten sein sollen. Standardwert: none.
• javaScriptTypeFnGcsPath : Der Cloud Storage-Pfad zur JavaScript-UDF-Quelle für eine Funktion, die angibt, dass „_index“-Metadaten im Dokument in der Bulk-Anfrage enthalten sein sollen.Standard: none.
• javaScriptTypeFnName : UDF-JavaScript-Funktionsname für eine Funktion, die angibt, dass '_type'-Metadaten im Dokument in der Bulk-Anfrage enthalten sein sollen. Standardwert: none.
• javaScriptIsDeleteFnGcsPath : Der Cloud Storage-Pfad zur JavaScript-UDF-Quelle für eine Funktion, die bestimmt, ob das Dokument gelöscht statt eingefügt oder aktualisiert werden soll. Die Funktion sollte den Stringwert „true“ oder „false“ zurückgeben. Standardwert: none.
• javaScriptIsDeleteFnName : UDF-JavaScript-Funktionsname für eine Funktion, die bestimmt, ob das Dokument gelöscht statt eingefügt oder aktualisiert werden soll. Die Funktion sollte den Stringwert „true“ oder „false“ zurückgeben. Standardwert: none.
• usePartialUpdate : Gibt an, ob Teilaktualisierungen (Aktualisieren statt Erstellen oder Indexieren, Teildokumente sind zulässig) in Elasticsearch-Anfragen verwendet werden sollen. Standardwert: false
• bulkInsertMethod : Gibt an, ob "INDEX" (Indexieren, Upsert zulässig) oder "CREATE" (Erstellen, Fehler bei doppelter _id) in Bulk-Anfragen von Elasticsearch verwendet werden soll. Standardeinstellung: "CREATE".
• trustSelfSignedCerts : Gibt an, ob dem selbst signierten Zertifikat vertraut werden soll. Eine installierte Elasticsearch-Instanz hat möglicherweise ein selbst signiertes Zertifikat. Aktivieren Sie diese Option auf „True“, um die Validierung des SSL-Zertifikats zu umgehen. Standardwert ist False.
• disableCertificateValidation : Wenn "true", wird dem selbstsignierten SSL-Zertifikat vertraut. Eine Elasticsearch-Instanz hat möglicherweise ein selbstsigniertes Zertifikat. Setzen Sie diesen Parameter auf "true", um die Überprüfung des Zertifikats zu umgehen. Standardeinstellung: false.
• apiKeyKMSEncryptionKey : Der Cloud KMS-Schlüssel zum Entschlüsseln des API-Schlüssels. Dieser Parameter muss angegeben werden, wenn apiKeySource auf KMS gesetzt ist. Wenn dieser Parameter angegeben wird, sollte der apiKey-String verschlüsselt übergeben werden. Parameter mit dem KMS API-Verschlüsselungsendpunkt verschlüsseln Der Schlüssel muss das Format „projects/{gcp_project}/locations/{key_region}/keyRings/{key_ring}/cryptoKeys/{kms_key_name}“ haben. Siehe https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt (Beispiel: projects/your-project-id/locations/global/keyRings/your-keyring/cryptoKeys/your-key-name).
• apiKeySecretId : Secret Manager-ID für den apiKey. Dieser Parameter sollte angegeben werden, wenn die apiKeySource auf SECRET_MANAGER festgelegt ist. Muss im Format „projects/{project}/secrets/{secret}/versions/{secret_version}“ vorliegen. Beispiel: projects/your-project-id/secrets/your-secret/versions/your-secret-version.
• apiKeySource : Quelle des API-Schlüssels. Entweder PLAINTEXT, KMS oder SECRET_MANAGER. Dieser Parameter muss angegeben werden, wenn Secret Manager oder KMS verwendet wird. Wenn apiKeySource auf KMS festgelegt ist, müssen apiKeyKMSEncryptionKey und verschlüsselter apiKey angegeben werden. Wenn apiKeySource auf SECRET_MANAGER festgelegt ist, muss apiKeySecretId angegeben werden. Wenn apiKeySource auf PLAINTEXT festgelegt ist, muss apiKey bereitgestellt werden. Die Standardeinstellung ist PLAINTEXT.
• javascriptTextTransformGcsPath : Das Cloud Storage-Pfadmuster für den JavaScript-Code, der Ihre benutzerdefinierten Funktionen enthält. Beispiel: gs://Ihr-Bucket/Ihre-Funktion.js.
• javascriptTextTransformFunctionName : Der Name der Funktion, die aus Ihrer JavaScript-Datei aufgerufen werden soll. Verwenden Sie nur Buchstaben, Ziffern und Unterstriche. (Beispiel: "transform" oder "transform_udf1").

Benutzerdefinierte Funktionen

Diese Vorlage unterstützt benutzerdefinierte Funktionen (UDFs) an mehreren Stellen in der Pipeline, wie unten beschrieben. Weitere Informationen finden Sie unter Benutzerdefinierte Funktionen für Dataflow-Vorlagen erstellen.

Indexfunktion

Gibt den Index zurück, zu dem das Dokument gehört.

Vorlagenparameter:

• javaScriptIndexFnGcsPath: der Cloud Storage-URI der JavaScript-Datei.
• javaScriptIndexFnName: Der Name der JavaScript-Funktion.

Funktionsspezifikation:

• Eingabe: Das Elasticsearch-Dokument, serialisiert als JSON-String.
• Ausgabe: Der Wert des Metadatenfelds _index des Dokuments.

Dokument-ID-Funktion

Gibt die Dokument-ID zurück

Vorlagenparameter:

• javaScriptIdFnGcsPath: der Cloud Storage-URI der JavaScript-Datei.
• javaScriptIdFnName: Der Name der JavaScript-Funktion.

Funktionsspezifikation:

• Eingabe: Das Elasticsearch-Dokument, serialisiert als JSON-String.
• Ausgabe: Der Wert des Metadatenfelds _id des Dokuments.

Funktion zum Löschen von Dokumenten

Gibt an, ob ein Dokument gelöscht werden soll. Um diese Funktion zu verwenden, legen Sie den Modus für Bulk-Einfügung auf INDEX fest und stellen eine Dokument-ID-Funktion bereit.

Vorlagenparameter:

• javaScriptIsDeleteFnGcsPath: der Cloud Storage-URI der JavaScript-Datei.
• javaScriptIsDeleteFnName: Der Name der JavaScript-Funktion.

Funktionsspezifikation:

• Eingabe: Das Elasticsearch-Dokument, serialisiert als JSON-String.
• Ausgabe: Geben Sie den String "true" zurück, um das Dokument zu löschen, oder "false", um das Dokument zu aktualisieren.

Funktion für Zuordnungstyp

Gibt den Zuordnungstyp des Dokuments zurück.

Achtung: Zuordnungstypen wurden in Elasticsearch 8.0 entfernt.

Vorlagenparameter:

• javaScriptTypeFnGcsPath: der Cloud Storage-URI der JavaScript-Datei.
• javaScriptTypeFnName: Der Name der JavaScript-Funktion.

Funktionsspezifikation:

• Eingabe: Das Elasticsearch-Dokument, serialisiert als JSON-String.
• Ausgabe: Der Wert des Metadatenfelds _type des Dokuments.

Führen Sie die Vorlage aus.

Console

1. Rufen Sie die Dataflow-Seite Job aus Vorlage erstellen auf.
Zur Seite "Job aus Vorlage erstellen“
2. Geben Sie im Feld Jobname einen eindeutigen Jobnamen ein.
3. Optional: Wählen Sie für Regionaler Endpunkt einen Wert aus dem Drop-down-Menü aus. Die Standardregion ist us-central1.

   Eine Liste der Regionen, in denen Sie einen Dataflow-Job ausführen können, finden Sie unter Dataflow-Standorte.

4. Wählen Sie im Drop-down-Menü Dataflow-Vorlage die Option the BigQuery to Elasticsearch templateaus.
5. Geben Sie Ihre Parameterwerte in die Parameterfelder ein.
6. Klicken Sie auf Job ausführen.

gcloud

Hinweis: Um flex-Vorlagen mit dem Google Cloud CLI auszuführen, benötigen Sie das Google Cloud CLI in der Version 284.0.0 oder höher.

Führen Sie die Vorlage in der Shell oder im Terminal aus:

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

Ersetzen Sie Folgendes:

• PROJECT_ID: die ID des Google Cloud-Projekts, in dem Sie den Dataflow-Job ausführen möchten
• JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
• REGION_NAME: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
• VERSION: Die Version der Vorlage, die Sie verwenden möchten

  Sie können die folgenden Werte verwenden:

  • latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
  • Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
  Achtung: Die neueste Version der Vorlagen wird möglicherweise mit funktionsgefährdenden Änderungen aktualisiert. In Ihren Produktionsumgebungen sollten Vorlagen verwendet werden, die sich im aktuellsten datierten übergeordneten Ordner befinden, um zu verhindern, dass diese funktionsgefährdenden Änderungen Ihre Produktionsworkflows beeinträchtigen.
• INPUT_TABLE_SPEC: ist der BigQuery-Tabellenname
• CONNECTION_URL: ist die Elasticsearch-URL
• APIKEY: ist der base64-codierte API-Schlüssel für die Authentifizierung
• INDEX: ist ihr Elasticsearch-Index

API

Senden Sie eine HTTP-POST-Anfrage, um die Vorlage mithilfe der REST API auszuführen. Weitere Informationen zur API und ihren Autorisierungsbereichen finden Sie unter 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",
   }
}

Ersetzen Sie Folgendes:

• PROJECT_ID: die ID des Google Cloud-Projekts, in dem Sie den Dataflow-Job ausführen möchten
• JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
• LOCATION: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
• VERSION: Die Version der Vorlage, die Sie verwenden möchten

  Sie können die folgenden Werte verwenden:

  • latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
  • Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
  Achtung: Die neueste Version der Vorlagen wird möglicherweise mit funktionsgefährdenden Änderungen aktualisiert. In Ihren Produktionsumgebungen sollten Vorlagen verwendet werden, die sich im aktuellsten datierten übergeordneten Ordner befinden, um zu verhindern, dass diese funktionsgefährdenden Änderungen Ihre Produktionsworkflows beeinträchtigen.
• INPUT_TABLE_SPEC: ist der BigQuery-Tabellenname
• CONNECTION_URL: ist die Elasticsearch-URL
• APIKEY: ist der base64-codierte API-Schlüssel für die Authentifizierung
• INDEX: ist ihr Elasticsearch-Index

Quellcode der Vorlage

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

Nächste Schritte

• Dataflow-Vorlagen
• Sehen Sie sich die Liste der von Google bereitgestellten Vorlagen an.