‫Google משתמשת בטכנולוגיית AI כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

תבנית BigQuery ל-Elasticsearch

תבנית BigQuery to Elasticsearch היא צינור להעברת נתונים באצוות, שמעביר נתונים מטבלה ב-BigQuery ל-Elasticsearch כמסמכים. התבנית יכולה לקרוא את כל הטבלה או לקרוא רשומות ספציפיות באמצעות שאילתה שסופקה.

הדרישות לגבי צינורות עיבוד נתונים

טבלת המקור ב-BigQuery צריכה להתקיים.
מארח Elasticsearch ב Google Cloud מופע או ב-Elastic Cloud עם Elasticsearch מגרסה 7.0 ואילך. צריכה להיות גישה אליה ממכונות העובדים של Dataflow.

פרמטרים של תבניות

פרמטרים נדרשים

‫connectionUrl: כתובת ה-URL של Elasticsearch בפורמט https://hostname:[port]. אם משתמשים ב-Elastic Cloud, צריך לציין את CloudID. לדוגמה, https://elasticsearch-host:9200.
‫apiKey: מפתח ה-API בקידוד Base64 שמשמש לאימות.
‫index: האינדקס של Elasticsearch שאליו מונפקות הבקשות. לדוגמה, my-index.

פרמטרים אופציונליים

‫inputTableSpec: הטבלה ב-BigQuery שממנה קוראים. אם מציינים את inputTableSpec, התבנית קוראת את הנתונים ישירות מאחסון BigQuery באמצעות BigQuery Storage Read API‏ (https://cloud.google.com/bigquery/docs/reference/storage). מידע על מגבלות ב-Storage Read API זמין בכתובת https://cloud.google.com/bigquery/docs/reference/storage#limitations. חובה לציין inputTableSpec או query. אם מגדירים את שני הפרמטרים, התבנית משתמשת בפרמטר query. לדוגמה, <BIGQUERY_PROJECT>:<DATASET_NAME>.<INPUT_TABLE>.
‫outputDeadletterTable: הטבלה ב-BigQuery להודעות שלא הצליחו להגיע לטבלת הפלט. אם טבלה לא קיימת, היא נוצרת במהלך הפעלת צינור הנתונים. אם לא מציינים ערך, המערכת משתמשת בערך <outputTableSpec>_error_records. לדוגמה, <PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE>.
‫query: שאילתת ה-SQL שמשמשת לקריאת נתונים מ-BigQuery. אם מערך הנתונים ב-BigQuery נמצא בפרויקט אחר ממשימת Dataflow, צריך לציין את השם המלא של מערך הנתונים בשאילתת ה-SQL, למשל: <PROJECT_ID>.<DATASET_NAME>.<TABLE_NAME>. כברירת מחדל, הפרמטר query משתמש ב-GoogleSQL (https://cloud.google.com/bigquery/docs/introduction-sql), אלא אם useLegacySql הוא true. חובה לציין inputTableSpec או query. אם מגדירים את שני הפרמטרים, התבנית משתמשת בפרמטר query. לדוגמה, select * from sampledb.sample_table.
‫useLegacySql: מגדירים את הערך ל-true כדי להשתמש ב-SQL מדור קודם. הפרמטר הזה רלוונטי רק כשמשתמשים בפרמטר query. ברירת המחדל היא false.
‫queryLocation: נדרש כשקוראים מתצוגה מורשית ללא הרשאה לטבלה הבסיסית. לדוגמה, US.
‫queryTempDataset: באמצעות האפשרות הזו, אפשר להגדיר מערך נתונים קיים כדי ליצור את הטבלה הזמנית לאחסון תוצאות השאילתה. לדוגמה, temp_dataset.
‫KMSEncryptionKey: אם קוראים מ-BigQuery באמצעות מקור שאילתה, משתמשים במפתח Cloud KMS הזה כדי להצפין טבלאות זמניות שנוצרו. לדוגמה, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
‫elasticsearchUsername: שם המשתמש ב-Elasticsearch לאימות. אם מציינים את הערך של apiKey, המערכת מתעלמת ממנו.
‫elasticsearchPassword: סיסמת Elasticsearch לאימות. אם מציינים את הערך של apiKey, המערכת מתעלמת ממנו.
‫batchSize: גודל האצווה במספר המסמכים. ברירת המחדל היא 1000.
‫batchSizeBytes: גודל הקבוצה במספר בייטים. ברירת המחדל היא 5242880 (5MB).
‫maxRetryAttempts: המספר המקסימלי של ניסיונות חוזרים. הערך חייב להיות גדול מאפס. ברירת המחדל היא 5.
‫maxRetryDuration: משך הזמן המקסימלי לניסיון חוזר, באלפיות השנייה. הערך חייב להיות גדול מאפס. ברירת המחדל היא 60000 (דקה אחת).
‫propertyAsIndex: המאפיין במסמך שמתווסף לאינדקס, שהערך שלו מציין מטא-נתונים של _index שייכללו במסמך בבקשות בכמות גדולה. מקבל עדיפות על פני UDF של _index. ברירת המחדל היא none.
‫javaScriptIndexFnGcsPath: הנתיב ב-Cloud Storage למקור של פונקציית JavaScript UDF, לפונקציה שמציינת מטא-נתונים של _index שיש לכלול במסמך בבקשות בכמות גדולה. ברירת המחדל היא none.
‫javaScriptIndexFnName: השם של פונקציית ה-JavaScript של ה-UDF שמציינת מטא-נתונים של _index שיש לכלול במסמך בבקשות בכמות גדולה. ברירת המחדל היא none.
‫propertyAsId: מאפיין במסמך שמתווסף לאינדקס, שהערך שלו מציין מטא-נתונים שייכללו עם המסמך בבקשות בכמות גדולה._id מקבל עדיפות על פני UDF של _id. ברירת המחדל היא none.
‫javaScriptIdFnGcsPath: הנתיב ב-Cloud Storage למקור של פונקציית ה-UDF ב-JavaScript, שמציינת מטא-נתונים של _id שיש לכלול במסמך בבקשות בכמות גדולה. ברירת המחדל היא none.
‫javaScriptIdFnName: השם של פונקציית JavaScript UDF שמציינת את המטא-נתונים של _id שייכללו במסמך בבקשות בכמות גדולה. ברירת המחדל היא none.
‫javaScriptTypeFnGcsPath: הנתיב ב-Cloud Storage למקור של פונקציית JavaScript UDF שמציינת מטא-נתונים של _type שיש לכלול במסמכים בבקשות בכמות גדולה. ברירת המחדל היא none.
‫javaScriptTypeFnName: השם של פונקציית ה-UDF JavaScript שמציינת את המטא-נתונים של _type שצריך לכלול במסמך בבקשות בכמות גדולה. ברירת המחדל היא none.
‫javaScriptIsDeleteFnGcsPath: הנתיב ב-Cloud Storage למקור של פונקציית ה-JavaScript UDF שקובעת אם למחוק את המסמך במקום להוסיף או לעדכן אותו. הפונקציה מחזירה מחרוזת עם הערך true או false. ברירת המחדל היא none.
‫javaScriptIsDeleteFnName: השם של פונקציית ה-JavaScript של ה-UDF שקובעת אם למחוק את המסמך במקום להוסיף או לעדכן אותו. הפונקציה מחזירה מחרוזת עם הערך true או false. ברירת המחדל היא none.
‫usePartialUpdate: האם להשתמש בעדכונים חלקיים (עדכון במקום יצירה או הוספה לאינדקס, שמאפשרים מסמכים חלקיים) בבקשות Elasticsearch. ברירת המחדל היא false.
‫bulkInsertMethod: האם להשתמש ב-INDEX (index, מאפשר עדכונים) או ב-CREATE (create, מחזיר שגיאה אם יש כפילות של _id) עם בקשות בכמות גדולה של Elasticsearch. ברירת המחדל היא CREATE.
‫trustSelfSignedCerts: האם לתת אמון באישור בחתימה עצמית או לא. יכול להיות שמופע Elasticsearch שהותקן כולל אישור בחתימה עצמית. כדי לעקוף את האימות של אישור ה-SSL, צריך להגדיר את הערך הזה כ-true. (ברירת המחדל: false).
‫disableCertificateValidation: אם הערך הוא true, המערכת בוטחת באישור ה-SSL בחתימה עצמית. יכול להיות שלמופע Elasticsearch יש אישור בחתימה עצמית. כדי לעקוף את האימות של האישור, מגדירים את הפרמטר הזה לערך true. ברירת המחדל היא false.
‫apiKeyKMSEncryptionKey: מפתח Cloud KMS לפענוח מפתח ה-API. חובה לכלול את הפרמטר הזה אם הערך של apiKeySource הוא KMS. אם הפרמטר הזה מסופק, צריך להעביר מחרוזת apiKey מוצפנת. הצפנת פרמטרים באמצעות נקודת הקצה להצפנה של KMS API. למפתח, משתמשים בפורמט projects/<PROJECT_ID>/locations/<KEY_REGION>/keyRings/<KEY_RING>/cryptoKeys/<KMS_KEY_NAME>. למידע נוסף: https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt לדוגמה, projects/your-project-id/locations/global/keyRings/your-keyring/cryptoKeys/your-key-name.
‫apiKeySecretId: המזהה הסודי ב-Secret Manager של מפתח ה-API. אם הערך של apiKeySource הוא SECRET_MANAGER, צריך לספק את הפרמטר הזה. משתמשים בפורמט projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. For example, projects/your-project-id/secrets/your-secret/versions/your-secret-version`.
‫apiKeySource: המקור של מפתח ה-API. הערכים המותרים הם PLAINTEXT, ‏ KMS או SECRET_MANAGER. חובה להשתמש בפרמטר הזה כשמשתמשים ב-Secret Manager או ב-KMS. אם הערך של apiKeySource הוא KMS, צריך לספק את apiKeyKMSEncryptionKey ואת מפתח ה-API המוצפן. אם הערך של apiKeySource הוא SECRET_MANAGER, צריך לספק את הערך של apiKeySecretId. אם הערך של apiKeySource הוא PLAINTEXT, צריך לספק את הערך של apiKey. ברירת המחדל היא: PLAINTEXT.
‫socketTimeout: אם מוגדר, מחליף את ברירת המחדל של הזמן הקצוב לתפוגה לניסיון חוזר ואת ברירת המחדל של הזמן הקצוב לתפוגה של socket (30,000 אלפיות שנייה) ב-Elastic RestClient.
‫javascriptTextTransformGcsPath: ה-URI של Cloud Storage של קובץ ה-‎ .js שמגדיר את הפונקציה בהגדרת המשתמש (UDF) ב-JavaScript שבה רוצים להשתמש. לדוגמה, gs://my-bucket/my-udfs/my_file.js.
‫javascriptTextTransformFunctionName: השם של פונקציית JavaScript בהגדרת המשתמש (UDF) שבה רוצים להשתמש. לדוגמה, אם קוד פונקציית ה-JavaScript הוא myTransform(inJson) { /*...do stuff...*/ }, אז שם הפונקציה הוא myTransform. דוגמאות ל-UDF ב-JavaScript זמינות במאמר UDF Examples (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples).

פונקציות בהגדרת המשתמש

התבנית הזו תומכת בפונקציות בהגדרת המשתמש (UDF) בכמה נקודות בצינור, כפי שמתואר בהמשך. למידע נוסף, ראו יצירת פונקציות מוגדרות על ידי המשתמש לתבניות Dataflow.

פונקציית אינדקס

הפונקציה מחזירה את האינדקס שאליו המסמך שייך.

פרמטרים של תבנית:

‫javaScriptIndexFnGcsPath: ה-URI של Cloud Storage של קובץ ה-JavaScript.
‫javaScriptIndexFnName: השם של פונקציית JavaScript.

מפרט הפונקציה:

קלט: מסמך Elasticsearch שעבר סריאליזציה כמחרוזת JSON.
‫Output: הערך של שדה המטא-נתונים _index במסמך.

הפונקציה Document ID

הפונקציה מחזירה את מזהה המסמך.

פרמטרים של תבנית:

‫javaScriptIdFnGcsPath: ה-URI של קובץ ה-JavaScript ב-Cloud Storage.
‫javaScriptIdFnName: השם של פונקציית JavaScript.

מפרט הפונקציה:

קלט: מסמך Elasticsearch שעבר סריאליזציה כמחרוזת JSON.
‫Output: הערך של שדה המטא-נתונים _id במסמך.

פונקציה למחיקת מסמכים

מציינים אם למחוק מסמך. כדי להשתמש בפונקציה הזו, צריך להגדיר את מצב ההוספה בכמות גדולה ל-INDEX ולספק פונקציית מזהה מסמך.

פרמטרים של תבנית:

‫javaScriptIsDeleteFnGcsPath: ה-URI של קובץ ה-JavaScript ב-Cloud Storage.
‫javaScriptIsDeleteFnName: השם של פונקציית JavaScript.

מפרט הפונקציה:

קלט: מסמך Elasticsearch שעבר סריאליזציה כמחרוזת JSON.
פלט: מחזירה את המחרוזת "true" כדי למחוק את המסמך, או את המחרוזת "false" כדי לעדכן או להוסיף (upsert) את המסמך.

פונקציית מיפוי סוג

מחזירה את סוג המיפוי של המסמך.

פרמטרים של תבנית:

‫javaScriptTypeFnGcsPath: ה-URI של Cloud Storage של קובץ ה-JavaScript.
‫javaScriptTypeFnName: השם של פונקציית JavaScript.

מפרט הפונקציה:

קלט: מסמך Elasticsearch שעבר סריאליזציה כמחרוזת JSON.
‫Output: הערך של שדה המטא-נתונים _type במסמך.

הרצת התבנית

המסוף

עוברים לדף Create job from template (יצירת משימה מתבנית) ב-Dataflow.

כניסה לדף Create job from template

בשדה שם המשימה, מזינים שם ייחודי למשימה.
אופציונלי: בשדה Regional endpoint (נקודת קצה אזורית), בוחרים ערך מהתפריט הנפתח. אזור ברירת המחדל הוא us-central1.
רשימת האזורים שבהם אפשר להריץ משימת Dataflow מופיעה במאמר בנושא מיקומי Dataflow.
בתפריט הנפתח Dataflow template (תבנית Dataflow), בוחרים בתבנית BigQuery to Elasticsearch (מ-BigQuery אל Elasticsearch).
בשדות הפרמטרים שמופיעים, מזינים את ערכי הפרמטרים.
לוחצים על הפעלת העבודה.

gcloud

במעטפת או בטרמינל, מריצים את התבנית:

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

מחליפים את מה שכתוב בשדות הבאים:

‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
‫REGION_NAME: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
‫VERSION: הגרסה של התבנית שבה רוצים להשתמש
אפשר להשתמש בערכים הבאים:
- ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית האב ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
- שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
‫INPUT_TABLE_SPEC: שם הטבלה ב-BigQuery.
‫CONNECTION_URL: כתובת ה-URL של Elasticsearch.
‫APIKEY: מפתח ה-API שלכם בקידוד base64 לאימות.
‫INDEX: האינדקס של Elasticsearch.

API

כדי להריץ את התבנית באמצעות API בארכיטקטורת REST, שולחים בקשת HTTP POST. מידע נוסף על ה-API ועל היקפי ההרשאות שלו זמין במאמר 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/",
   }
}

מחליפים את מה שכתוב בשדות הבאים:

‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
‫LOCATION: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
‫VERSION: הגרסה של התבנית שבה רוצים להשתמש
אפשר להשתמש בערכים הבאים:
- ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית האב ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
- שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
‫INPUT_TABLE_SPEC: שם הטבלה ב-BigQuery.
‫CONNECTION_URL: כתובת ה-URL של Elasticsearch.
‫APIKEY: מפתח ה-API שלכם בקידוד base64 לאימות.
‫INDEX: האינדקס של Elasticsearch.

קוד המקור של התבנית

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

המאמרים הבאים

מידע נוסף על תבניות Dataflow
כאן אפשר לעיין ברשימת התבניות ש-Google סיפקה.

תבנית BigQuery ל-Elasticsearch קל לארגן דפים בעזרת אוספים אפשר לשמור ולסווג תוכן על סמך ההעדפות שלך.

הדרישות לגבי צינורות עיבוד נתונים

פרמטרים של תבניות

פרמטרים נדרשים

פרמטרים אופציונליים

פונקציות בהגדרת המשתמש

פונקציית אינדקס

הפונקציה Document ID

פונקציה למחיקת מסמכים

פונקציית מיפוי סוג

הרצת התבנית

המסוף

gcloud

API

קוד המקור של התבנית

Java

המאמרים הבאים

תבנית BigQuery ל-Elasticsearch