תבנית Cloud Storage ל-Elasticsearch

בטא

השימוש בתכונה הזו כפוף לתנאי השימוש במוצרים בגרסת טרום-GA, שמפורטים בחלק "תנאי שירות הכלליים" בתנאים הספציפיים של השירות. השימוש בתכונות בגרסת טרום-GA הוא "כפי שהן" (As is), ויכול להיות שהתמיכה בהן תהיה מוגבלת. מידע נוסף זמין בקטע תיאור שלבי ההשקה.

התבנית Cloud Storage to Elasticsearch היא צינור להעברת נתונים באצווה שקורא נתונים מקובצי CSV שמאוחסנים בקטגוריה של Cloud Storage וכותב את הנתונים ל-Elasticsearch כמסמכי JSON.

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

• הקטגוריה של Cloud Storage צריכה להתקיים.
• צריך שיהיה קיים מארח Elasticsearch במופע Google Cloud או ב-Elasticsearch Cloud שאפשר לגשת אליו מ-Dataflow.
• צריכה להיות קיימת טבלה ב-BigQuery לפלט של השגיאות.

סכימת CSV

אם קובצי ה-CSV מכילים כותרות, צריך להגדיר את פרמטר התבנית containsHeaders לערך true.

אחרת, יוצרים קובץ סכימת JSON שמתאר את הנתונים. מציינים את ה-URI של Cloud Storage של קובץ הסכימה בפרמטר התבנית jsonSchemaPath. בדוגמה הבאה מוצגת סכימת JSON:

[{"name":"id", "type":"text"}, {"name":"age", "type":"integer"}]

לחלופין, אפשר לספק פונקציה בהגדרת המשתמש (UDF) שמנתחת את טקסט ה-CSV ומפיקה מסמכי Elasticsearch.

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

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

• ‫deadletterTable: טבלת ה-dead-letter ב-BigQuery שאליה יישלחו נתונים שלא הצליחו להיכנס. לדוגמה, your-project:your-dataset.your-table-name.
• ‫inputFileSpec: תבנית הקובץ ב-Cloud Storage לחיפוש קובצי CSV. לדוגמה, gs://mybucket/test-*.csv.
• ‫connectionUrl: כתובת ה-URL של Elasticsearch בפורמט https://hostname:[port]. אם משתמשים ב-Elastic Cloud, צריך לציין את CloudID. לדוגמה, https://elasticsearch-host:9200.
• ‫apiKey: מפתח ה-API בקידוד Base64 שמשמש לאימות.
• ‫index: האינדקס של Elasticsearch שאליו מונפקות הבקשות. לדוגמה, my-index.

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

• ‫inputFormat: הפורמט של קובץ הקלט. ברירת המחדל היא CSV.
• ‫containsHeaders: קובצי ה-CSV של הקלט מכילים רשומת כותרת (true/false). נדרש רק אם קוראים קובצי CSV. ברירת המחדל היא: false.
• ‫delimiter: תו ההפרדה בין העמודות בקובצי הטקסט של הקלט. ברירת מחדל: , לדוגמה, ,.
• ‫csvFormat: מפרט של פורמט CSV לשימוש בניתוח רשומות. ברירת המחדל היא: Default. פרטים נוספים זמינים בכתובת https://commons.apache.org/proper/commons-csv/apidocs/org/apache/commons/csv/CSVFormat.html. השם חייב להיות זהה בדיוק לפורמט שמופיע בכתובת: https://commons.apache.org/proper/commons-csv/apidocs/org/apache/commons/csv/CSVFormat.Predefined.html.
• ‫jsonSchemaPath: הנתיב לסכימת ה-JSON. ברירת המחדל היא null. לדוגמה, gs://path/to/schema.
• ‫largeNumFiles: מגדירים את הערך true אם מספר הקבצים הוא עשרות אלפים. ברירת המחדל היא false.
• ‫csvFileEncoding: פורמט קידוד התווים של קובץ ה-CSV. הערכים המותרים הם US-ASCII,‏ ISO-8859-1,‏ UTF-8 ו-UTF-16. ברירת המחדל היא UTF-8.
• ‫logDetailedCsvConversionErrors: מגדירים את הערך true כדי להפעיל רישום מפורט של שגיאות ביומן אם ניתוח קובץ ה-CSV נכשל. שימו לב שהפעולה הזו עשויה לחשוף מידע אישי רגיש ביומנים (לדוגמה, אם קובץ ה-CSV מכיל סיסמאות). ברירת מחדל: false.
• ‫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.

פונקציית שינוי הטקסט

הופך את נתוני ה-CSV למסמך Elasticsearch.

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

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

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

• קלט: שורה אחת מקובץ CSV של קלט.
• פלט: מסמך JSON שהומר למחרוזת להוספה ל-Elasticsearch.

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

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

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

• ‫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) את המסמך.

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

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

שימו לב: מיפויי סוגים הוסרו ב-Elasticsearch 8.0.

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

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

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

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

הרצת התבנית

המסוף

1. עוברים לדף Create job from template (יצירת משימה מתבנית) ב-Dataflow.
כניסה לדף Create job from template
2. בשדה שם המשימה, מזינים שם ייחודי למשימה.
3. אופציונלי: בשדה Regional endpoint (נקודת קצה אזורית), בוחרים ערך מהתפריט הנפתח. אזור ברירת המחדל הוא us-central1.

   רשימת האזורים שבהם אפשר להריץ משימת Dataflow מופיעה במאמר בנושא מיקומי Dataflow.

4. בתפריט הנפתח Dataflow template (תבנית Dataflow), בוחרים בתבנית Cloud Storage to Elasticsearch (מ-Cloud Storage אל Elasticsearch).
5. בשדות הפרמטרים שמופיעים, מזינים את ערכי הפרמטרים.
6. לוחצים על הפעלת העבודה.

gcloud

הערה: כדי להשתמש ב-Google Cloud CLI להפעלת תבניות flex צריך להשתמש בגרסה 284.0.0 או בגרסה מאוחרת יותר של Google Cloud CLI.

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

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 \
inputFileSpec=INPUT_FILE_SPEC,\
connectionUrl=CONNECTION_URL,\
apiKey=APIKEY,\
index=INDEX,\
deadletterTable=DEADLETTER_TABLE,\

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

• ‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

  אפשר להשתמש בערכים הבאים:

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

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": {
          "inputFileSpec": "INPUT_FILE_SPEC",
          "connectionUrl": "CONNECTION_URL",
          "apiKey": "APIKEY",
          "index": "INDEX",
          "deadletterTable": "DEADLETTER_TABLE"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/",
   }
}

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

• ‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

  אפשר להשתמש בערכים הבאים:

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

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

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 static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;

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.coders.FailsafeElementCoder;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.elasticsearch.options.GCSToElasticsearchOptions;
import com.google.cloud.teleport.v2.elasticsearch.transforms.WriteToElasticsearch;
import com.google.cloud.teleport.v2.transforms.CsvConverters;
import com.google.cloud.teleport.v2.transforms.ErrorConverters.WriteStringMessageErrors;
import com.google.cloud.teleport.v2.utils.SchemaUtils;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import com.google.common.base.Strings;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.CoderRegistry;
import org.apache.beam.sdk.coders.NullableCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.WithTimestamps;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link GCSToElasticsearch} pipeline exports data from one or more CSV files in Cloud Storage
 * to Elasticsearch.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-elasticsearch/README_GCS_to_Elasticsearch.md">README</a>
 * for instructions on how to use or modify this template.
 */
@MultiTemplate({
  @Template(
      name = "GCS_to_Elasticsearch",
      category = TemplateCategory.BATCH,
      displayName = "Cloud Storage to Elasticsearch",
      description = {
        "The Cloud Storage to Elasticsearch template is a batch pipeline that reads data from CSV files stored in a Cloud Storage bucket and writes the data into Elasticsearch as JSON documents.",
        "If the CSV files contain headers, set the <code>containsHeaders</code> template parameter to <code>true</code>.\n"
            + "Otherwise, create a JSON schema file that describes the data. Specify the Cloud Storage URI of the schema file in the jsonSchemaPath template parameter. "
            + "The following example shows a JSON schema:\n"
            + "<code>[{\"name\":\"id\", \"type\":\"text\"}, {\"name\":\"age\", \"type\":\"integer\"}]</code>\n"
            + "Alternatively, you can provide a Javascript user-defined function (UDF) that parses the CSV text and outputs Elasticsearch documents."
      },
      optionsClass = GCSToElasticsearchOptions.class,
      skipOptions = {
        "javascriptTextTransformReloadIntervalMinutes",
        "pythonExternalTextTransformGcsPath",
        "pythonExternalTextTransformFunctionName"
      },
      flexContainerName = "googlecloud-to-elasticsearch",
      documentation =
          "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-storage-to-elasticsearch",
      contactInformation = "https://cloud.google.com/support",
      preview = true,
      requirements = {
        "The Cloud Storage bucket must exist.",
        "A Elasticsearch host on a Google Cloud instance or on Elasticsearch Cloud that is accessible from Dataflow must exist.",
        "A BigQuery table for error output must exist."
      }),
  @Template(
      name = "GCS_to_Elasticsearch_Xlang",
      category = TemplateCategory.BATCH,
      displayName = "Cloud Storage to Elasticsearch with Python UDFs",
      type = Template.TemplateType.XLANG,
      description = {
        "The Cloud Storage to Elasticsearch template is a batch pipeline that reads data from CSV files stored in a Cloud Storage bucket and writes the data into Elasticsearch as JSON documents.",
        "If the CSV files contain headers, set the <code>containsHeaders</code> template parameter to <code>true</code>.\n"
            + "Otherwise, create a JSON schema file that describes the data. Specify the Cloud Storage URI of the schema file in the jsonSchemaPath template parameter. "
            + "The following example shows a JSON schema:\n"
            + "<code>[{\"name\":\"id\", \"type\":\"text\"}, {\"name\":\"age\", \"type\":\"integer\"}]</code>\n"
            + "Alternatively, you can provide a Python user-defined function (UDF) that parses the CSV text and outputs Elasticsearch documents."
      },
      optionsClass = GCSToElasticsearchOptions.class,
      skipOptions = {
        "javascriptTextTransformGcsPath",
        "javascriptTextTransformFunctionName",
        "javascriptTextTransformReloadIntervalMinutes"
      },
      flexContainerName = "googlecloud-to-elasticsearch-xlang",
      documentation =
          "https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-storage-to-elasticsearch",
      contactInformation = "https://cloud.google.com/support",
      preview = true,
      requirements = {
        "The Cloud Storage bucket must exist.",
        "A Elasticsearch host on a Google Cloud instance or on Elasticsearch Cloud that is accessible from Dataflow must exist.",
        "A BigQuery table for error output must exist."
      })
})
public class GCSToElasticsearch {

  /** The tag for the headers of the CSV if required. */
  static final TupleTag<String> CSV_HEADERS = new TupleTag<String>() {};

  /** The tag for the lines of the CSV. */
  static final TupleTag<String> CSV_LINES = new TupleTag<String>() {};

  /** The tag for the dead-letter output of the UDF. */
  static final TupleTag<FailsafeElement<String, String>> PROCESSING_DEADLETTER_OUT =
      new TupleTag<FailsafeElement<String, String>>() {};

  /** The tag for the main output for the UDF. */
  static final TupleTag<FailsafeElement<String, String>> PROCESSING_OUT =
      new TupleTag<FailsafeElement<String, String>>() {};

  /* Logger for class. */
  private static final Logger LOG = LoggerFactory.getLogger(GCSToElasticsearch.class);

  /** String/String Coder for FailsafeElement. */
  private static final FailsafeElementCoder<String, String> FAILSAFE_ELEMENT_CODER =
      FailsafeElementCoder.of(
          NullableCoder.of(StringUtf8Coder.of()), NullableCoder.of(StringUtf8Coder.of()));

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    GCSToElasticsearchOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(GCSToElasticsearchOptions.class);

    run(options);
  }

  /**
   * Runs the pipeline to completion with the specified options.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  private static PipelineResult run(GCSToElasticsearchOptions options) {
    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    // Register the coder for pipeline
    CoderRegistry coderRegistry = pipeline.getCoderRegistry();
    coderRegistry.registerCoderForType(
        FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor(), FAILSAFE_ELEMENT_CODER);

    // Throw error if containsHeaders is true and a schema or Udf is also set.
    if (options.getContainsHeaders()) {
      checkArgument(
          options.getJavascriptTextTransformGcsPath() == null
              && options.getJsonSchemaPath() == null
              && options.getPythonExternalTextTransformGcsPath() == null,
          "Cannot parse file containing headers with UDF or Json schema.");
    }

    // Throw error if only one retry configuration parameter is set.
    checkArgument(
        (options.getMaxRetryAttempts() == null && options.getMaxRetryDuration() == null)
            || (options.getMaxRetryAttempts() != null && options.getMaxRetryDuration() != null),
        "To specify retry configuration both max attempts and max duration must be set.");

    // Throw error if both Javascript UDF and Python UDF are set. We can only apply one or the
    // other.
    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.");
    }

    /*
     * Steps: 1) Read records from CSV(s) via {@link CsvConverters.ReadCsv}.
     *        2) Convert lines to JSON strings via {@link CsvConverters.LineToFailsafeJson}.
     *        3a) Write JSON strings as documents to Elasticsearch via {@link ElasticsearchIO}.
     *        3b) Write elements that failed processing to {@link org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO}.
     */
    PCollectionTuple readCsvLines =
        pipeline
            /*
             * Step 1: Read CSV file(s) from Cloud Storage using {@link CsvConverters.ReadCsv}.
             */
            .apply(
            "ReadCsv",
            CsvConverters.ReadCsv.newBuilder()
                .setCsvFormat(options.getCsvFormat())
                .setDelimiter(options.getDelimiter())
                .setHasHeaders(options.getContainsHeaders())
                .setInputFileSpec(options.getInputFileSpec())
                .setHeaderTag(CSV_HEADERS)
                .setLineTag(CSV_LINES)
                .setFileEncoding(options.getCsvFileEncoding())
                .build());
    /*
     * Step 2: Convert lines to Elasticsearch document.
     */
    CsvConverters.LineToFailsafeJson.Builder lineToFailsafeJsonBuilder =
        CsvConverters.LineToFailsafeJson.newBuilder()
            .setDelimiter(options.getDelimiter())
            .setJsonSchemaPath(options.getJsonSchemaPath())
            .setHeaderTag(CSV_HEADERS)
            .setLineTag(CSV_LINES)
            .setUdfOutputTag(PROCESSING_OUT)
            .setUdfDeadletterTag(PROCESSING_DEADLETTER_OUT);
    if (options.getPythonExternalTextTransformGcsPath() != null) {
      lineToFailsafeJsonBuilder
          .setPythonUdfFileSystemPath(options.getPythonExternalTextTransformGcsPath())
          .setPythonUdfFunctionName(options.getPythonExternalTextTransformFunctionName());
    } else {
      lineToFailsafeJsonBuilder
          .setJavascriptUdfFileSystemPath(options.getJavascriptTextTransformGcsPath())
          .setJavascriptUdfFunctionName(options.getJavascriptTextTransformFunctionName());
    }
    PCollectionTuple convertedCsvLines =
        readCsvLines.apply("ConvertLine", lineToFailsafeJsonBuilder.build());
    /*
     * Step 3a: Write elements that were successfully processed to Elasticsearch using {@link WriteToElasticsearch}.
     */
    convertedCsvLines
        .get(PROCESSING_OUT)
        .apply(
            "GetJsonDocuments",
            MapElements.into(TypeDescriptors.strings()).via(FailsafeElement::getPayload))
        .apply(
            "WriteToElasticsearch",
            WriteToElasticsearch.newBuilder()
                .setUserAgent("dataflow-gcs-to-elasticsearch-template/v2")
                .setOptions(options.as(GCSToElasticsearchOptions.class))
                .build());

    /*
     * Step 3b: Write elements that failed processing to deadletter table via {@link BigQueryIO}.
     */
    convertedCsvLines
        .get(PROCESSING_DEADLETTER_OUT)
        .apply(
            "AddTimestamps",
            WithTimestamps.of((FailsafeElement<String, String> failures) -> new Instant()))
        .apply(
            "WriteFailedElementsToBigQuery",
            WriteStringMessageErrors.newBuilder()
                .setErrorRecordsTable(options.getDeadletterTable())
                .setErrorRecordsTableSchema(SchemaUtils.DEADLETTER_SCHEMA)
                .build());

    return pipeline.run();
  }
}

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

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