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

תבנית Cloud Storage ל-Elasticsearch

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

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

הקטגוריה של Cloud Storage צריכה להתקיים.
צריך שיהיה קיים מארח Elasticsearch במופע של Google Cloud Platform או ב-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: המספר המקסימלי של ניסיונות חוזרים. הערך חייב להיות גדול מאפס. ברירת המחדל היא 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.

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

הופך את נתוני ה-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) את המסמך.

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

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

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

‫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 Cloud Storage 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/GCS_to_Elasticsearch \
    --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/GCS_to_Elasticsearch",
   }
}

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

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

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

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

סכימת CSV

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

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

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

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

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

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

הפונקציה Document ID

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

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

הפעלת התבנית

המסוף

gcloud

API

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

Java

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

תבנית Cloud Storage ל-Elasticsearch