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

תבנית BigQuery ל-Elasticsearch

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

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

טבלת המקור ב-BigQuery צריכה להתקיים.
מארח Elasticsearch במופע של Google Cloud Platform או ב-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: המספר המקסימלי של ניסיונות חוזרים. הערך חייב להיות גדול מאפס. ברירת המחדל היא no retries.
‫maxRetryDuration: משך הניסיון החוזר המקסימלי באלפיות השנייה. הערך חייב להיות גדול מאפס. ברירת המחדל היא no retries.
‫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: השם של פונקציית ה-UDF ב-JavaScript שמציינת את המטא-נתונים של _id שייכללו במסמך בבקשות בכמות גדולה. ברירת המחדל היא none.
‫javaScriptTypeFnGcsPath: הנתיב ב-Cloud Storage למקור של פונקציית JavaScript UDF שמציינת מטא-נתונים מסוג _type שיש לכלול במסמכים בבקשות בכמות גדולה. ברירת המחדל היא none.
‫javaScriptTypeFnName: השם של פונקציית ה-JavaScript של ה-UDF שמציינת את המטא-נתונים של _type שיש לכלול במסמך בבקשות בכמות גדולה. ברירת המחדל היא none.
‫javaScriptIsDeleteFnGcsPath: הנתיב ב-Cloud Storage למקור של פונקציית ה-UDF ב-JavaScript, שקובעת אם למחוק את המסמך במקום להוסיף או לעדכן אותו. הפונקציה מחזירה מחרוזת עם הערך true או false. ברירת המחדל היא none.
‫javaScriptIsDeleteFnName: השם של פונקציית ה-JavaScript של ה-UDF שקובעת אם למחוק את המסמך במקום להוסיף או לעדכן אותו. הפונקציה מחזירה מחרוזת עם הערך true או false. ברירת המחדל היא none.
‫usePartialUpdate: האם להשתמש בעדכונים חלקיים (עדכון במקום יצירה או הוספה לאינדקס, שמאפשרים מסמכים חלקיים) בבקשות Elasticsearch. ברירת המחדל היא false.
‫bulkInsertMethod: האם להשתמש ב-INDEX (index, allows upserts) או ב-CREATE (create, errors on duplicate _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: אם ההגדרה הזו מוגדרת, היא מחליפה את ברירת המחדל של הזמן הקצוב לתפוגה של הניסיון החוזר ואת ברירת המחדל של הזמן הקצוב לתפוגה של שקע (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 במסמך.

הפעלת התבנית

המסוף

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

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

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

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/BigQuery_to_Elasticsearch \
    --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/BigQuery_to_Elasticsearch",
   }
}

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

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

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

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

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

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

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

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

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

הפונקציה Document ID

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

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

הפעלת התבנית

המסוף

gcloud

API

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

Java

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

תבנית BigQuery ל-Elasticsearch