תבנית Pub/Sub ל-Elasticsearch

תבנית Pub/Sub to Elasticsearch היא צינור עיבוד נתונים (pipeline) של סטרימינג שקורא הודעות ממינוי Pub/Sub, מפעיל פונקציה בהגדרת המשתמש (UDF) וכותב אותן ל-Elasticsearch כמסמכים. תבנית Dataflow משתמשת בתכונה data streams של Elasticsearch כדי לאחסן נתונים של סדרות זמנים בכמה אינדקסים, ובו-זמנית מספקת לכם משאב יחיד עם שם לבקשות. מקורות נתונים מתאימים ליומנים, למדדים, למעקב ולנתונים אחרים שנוצרים באופן רציף ומאוחסנים ב-Pub/Sub.

התבנית יוצרת מקור נתונים בשם logs-gcp.DATASET-NAMESPACE, כאשר:

  • ‫DATASET הוא הערך של פרמטר התבנית dataset, או pubsub אם לא צוין.
  • ‫NAMESPACE הוא הערך של פרמטר התבנית namespace, או default אם לא צוין.

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

  • מינוי המקור ל-Pub/Sub חייב להתקיים, וההודעות צריכות להיות מקודדות בפורמט JSON תקין.
  • מארח Elasticsearch שאפשר להגיע אליו באופן ציבורי במופע Google Cloud או ב-Elastic Cloud עם Elasticsearch מגרסה 7.0 ומעלה. פרטים נוספים זמינים במאמר שילוב של Google Cloud עם Elastic.
  • נושא Pub/Sub לפלט של שגיאות.

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

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

  • ‫inputSubscription: מינוי ל-Pub/Sub לצריכת הקלט. לדוגמה, projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_NAME>.
  • ‫errorOutputTopic: נושא הפלט של Pub/Sub לפרסום רשומות שנכשלו, בפורמט projects/<PROJECT_ID>/topics/<TOPIC_NAME>.
  • ‫connectionUrl: כתובת ה-URL של Elasticsearch בפורמט https://hostname:[port]. אם משתמשים ב-Elastic Cloud, צריך לציין את CloudID. לדוגמה, https://elasticsearch-host:9200.
  • ‫apiKey: מפתח ה-API בקידוד Base64 שמשמש לאימות.

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

  • dataset: סוג היומנים שנשלחים באמצעות Pub/Sub, שעבורם יש לנו לוח בקרה מוכן לשימוש. ערכי סוגי היומנים המוכרים הם audit,‏ vpcflow ו-firewall. ברירת המחדל: pubsub.
  • ‫namespace: קיבוץ שרירותי, כמו סביבה (פיתוח, ייצור או QA), צוות או יחידה עסקית אסטרטגית. ברירת המחדל: default.
  • ‫elasticsearchTemplateVersion: מזהה גרסת התבנית של Dataflow, בדרך כלל מוגדר על ידי Google Cloud. ברירת המחדל היא: 1.0.0.
  • ‫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).
  • ‫javascriptTextTransformReloadIntervalMinutes: מציין את התדירות שבה יש לטעון מחדש את הפונקציה המוגדרת על ידי המשתמש, בדקות. אם הערך גדול מ-0, מערכת Dataflow בודקת מעת לעת את קובץ ה-UDF ב-Cloud Storage, ומטעינה מחדש את ה-UDF אם הקובץ משתנה. הפרמטר הזה מאפשר לכם לעדכן את ה-UDF בזמן שהצינור פועל, בלי שתצטרכו להפעיל מחדש את העבודה. אם הערך הוא 0, טעינה מחדש של פונקציות UDF מושבתת. ערך ברירת המחדל הוא 0.
  • ‫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, 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.

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

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

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

הופך את הודעת Pub/Sub למסמך Elasticsearch.

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

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

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

  • קלט: שדה הנתונים של הודעת Pub/Sub, שעבר סריאליזציה כמחרוזת JSON.
  • פלט: מסמך JSON שהומר למחרוזת להוספה ל-Elasticsearch.

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

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

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

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

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

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

פונקציית מזהה המסמך

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

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

  • ‫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 במסמך.

הרצת התבנית

המסוף

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

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

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

gcloud

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

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/PubSub_to_Elasticsearch_Flex \
    --parameters \
inputSubscription=SUBSCRIPTION_NAME,\
connectionUrl=CONNECTION_URL,\
dataset=DATASET,\
namespace=NAMESPACE,\
apiKey=APIKEY,\
errorOutputTopic=ERROR_OUTPUT_TOPIC

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

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

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

    • latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates/latest/
    • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית. אפשר למצוא את הגרסה הזו בתיקיית ההורה המתאימה עם התאריך בדלי – gs://dataflow-templates/
  • ‫ERROR_OUTPUT_TOPIC: נושא ה-Pub/Sub שלכם לפלט שגיאות
  • ‫SUBSCRIPTION_NAME: השם של המינוי שלכם ל-Pub/Sub
  • ‫CONNECTION_URL: כתובת ה-URL של Elasticsearch
  • ‫DATASET: סוג היומן
  • ‫NAMESPACE: מרחב השמות של מערך הנתונים
  • ‫APIKEY: מפתח ה-API שלכם בקידוד base64 לאימות

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": {
          "inputSubscription": "SUBSCRIPTION_NAME",
          "connectionUrl": "CONNECTION_URL",
          "dataset": "DATASET",
          "namespace": "NAMESPACE",
          "apiKey": "APIKEY",
          "errorOutputTopic": "ERROR_OUTPUT_TOPIC"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/PubSub_to_Elasticsearch_Flex",
   }
}

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

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

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

    • latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates/latest/
    • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית. אפשר למצוא את הגרסה הזו בתיקיית ההורה המתאימה עם התאריך בדלי – gs://dataflow-templates/
  • ‫ERROR_OUTPUT_TOPIC: נושא ה-Pub/Sub שלכם לפלט שגיאות
  • ‫SUBSCRIPTION_NAME: השם של המינוי שלכם ל-Pub/Sub
  • ‫CONNECTION_URL: כתובת ה-URL של Elasticsearch
  • ‫DATASET: סוג היומן
  • ‫NAMESPACE: מרחב השמות של מערך הנתונים
  • ‫APIKEY: מפתח ה-API שלכם בקידוד base64 לאימות

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