תבנית של Apache Kafka ל-Cloud Storage

התבנית Apache Kafka to Cloud Storage היא פייפליין להזרמת נתונים שקולט נתוני טקסט מ-שירות מנוהל של Google Cloud ל-Apache Kafka ומפיק את הרשומות ל-Cloud Storage.

אפשר גם להשתמש בתבנית Apache Kafka to Cloud Storage עם Kafka בניהול עצמי או חיצוני.

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

• קטגוריית הפלט של Cloud Storage חייבת להתקיים.
• שרת הברוקר של Apache Kafka צריך לפעול ולהיות נגיש ממכונות העובד של Dataflow.
• הנושאים של Apache Kafka צריכים להתקיים.

פורמט ההודעה ב-Kafka

התבנית הזו תומכת בקריאת הודעות מ-Kafka בפורמטים הבאים:

• JSON
• קידוד בינארי של Avro
• ‫Avro עם קידוד של Confluent Schema Registry

פורמט JSON

כדי לקרוא הודעות JSON, מגדירים את פרמטר התבנית messageFormat לערך "JSON".

קידוד בינארי של Avro

כדי לקרוא הודעות בינאריות בפורמט Avro, צריך להגדיר את פרמטרים התבנית הבאים:

• messageFormat: "AVRO_BINARY_ENCODING".
• ‫binaryAvroSchemaPath: המיקום של קובץ סכימת Avro ב-Cloud Storage. לדוגמה: gs://BUCKET_NAME/message-schema.avsc.

למידע נוסף על פורמט בינארי של Avro, אפשר לעיין במאמר בנושא קידוד בינארי במסמכי התיעוד של Apache Avro.

‫Avro מקודד של Confluent Schema Registry

כדי לקרוא הודעות ב-Avro שמקודדות ב-Confluent Schema Registry, מגדירים את פרמטרים התבנית הבאים:

• messageFormat: "AVRO_CONFLUENT_WIRE_FORMAT".

• ‫schemaFormat: אחד מהערכים הבאים:
  • ‫"SINGLE_SCHEMA_FILE": סכימת ההודעה מוגדרת בקובץ סכימת Avro. מציינים את המיקום של קובץ הסכימה ב-Cloud Storage בפרמטר confluentAvroSchemaPath.
  • ‫"SCHEMA_REGISTRY": ההודעות מקודדות באמצעות Confluent Schema Registry. מציינים את כתובת ה-URL של מופע Confluent Schema Registry בפרמטר schemaRegistryConnectionUrl, ומציינים את מצב האימות בפרמטר schemaRegistryAuthenticationMode.

מידע נוסף על הפורמט הזה זמין במאמר בנושא פורמט העברה במאמרי העזרה של Confluent.

פורמט קובץ הפלט

פורמט קובץ הפלט זהה לפורמט של הודעת הקלט של Kafka. לדוגמה, אם בוחרים באפשרות JSON לפורמט ההודעה של Kafka, קובצי JSON נכתבים לקטגוריה של Cloud Storage לפלט.

אימות

תבנית Apache Kafka to Cloud Storage תומכת באימות SASL/PLAIN בברוקרים של Kafka.

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

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

• ‫readBootstrapServerAndTopic: נושא Kafka לקריאת הקלט ממנו.
• ‫outputDirectory: הנתיב והקידומת של שם הקובץ לכתיבת קובצי פלט. חייב להסתיים בלוכסן. לדוגמה, gs://your-bucket/your-path/.
• ‫kafkaReadAuthenticationMode: מצב האימות לשימוש עם אשכול Kafka. משתמשים ב-NONE ללא אימות, ב-SASL_PLAIN לשם משתמש וסיסמה של SASL/PLAIN, ב-SASL_SCRAM_512 לאימות SASL_SCRAM_512 וב-TLS לאימות מבוסס-אישור. המאפיין APPLICATION_DEFAULT_CREDENTIALS צריך לשמש רק עבור אשכול של שירות מנוהל של Google Cloud ל-Apache Kafka. הוא מאפשר אימות באמצעות פרטי כניסה שמוגדרים כברירת מחדל באפליקציה.
• ‫messageFormat: הפורמט של הודעות Kafka לקריאה. הערכים הנתמכים הם AVRO_CONFLUENT_WIRE_FORMAT (Avro מקודד של Confluent Schema Registry),‏ AVRO_BINARY_ENCODING (Avro בינארי רגיל) ו-JSON. ברירת המחדל היא: AVRO_CONFLUENT_WIRE_FORMAT.
• ‫useBigQueryDLQ: אם הערך הוא true, הודעות שנכשלו ייכתבו ל-BigQuery עם מידע נוסף על השגיאה. ברירת המחדל היא false.

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

• ‫windowDuration: משך החלון או גודל החלון שבו הנתונים ייכתבו ל-Cloud Storage. הפורמטים המותרים הם: Ns (לשניות, לדוגמה: 5s), ‏ Nm (לדקות, לדוגמה: 12m), ‏ Nh (לשעות, לדוגמה: 2h). לדוגמה, 5m. ברירת המחדל היא: 5m.
• ‫outputFilenamePrefix: הקידומת שתוצב בכל קובץ של חלון. לדוגמה, output-. ברירת המחדל היא: output.
• ‫numShards: המספר המקסימלי של רסיסי פלט שנוצרים בזמן כתיבה. מספר גבוה יותר של רסיסים פירושו תפוקה גבוהה יותר לכתיבה ב-Cloud Storage, אבל עלולות להיות עלויות גבוהות יותר של צבירת נתונים ברסיסים כשמעבדים קבצים של פלט Cloud Storage. ערך ברירת המחדל נקבע על ידי Dataflow.
• ‫enableCommitOffsets: ביצוע Commit של אופסטים של הודעות שעברו עיבוד ל-Kafka. אם האפשרות הזו מופעלת, היא תצמצם את הפערים או את העיבוד הכפול של הודעות כשמפעילים מחדש את צינור העיבוד. חובה לציין את מזהה קבוצת הצרכנים. ברירת המחדל היא false.
• ‫consumerGroupId: המזהה הייחודי של קבוצת הצרכנים שאליה שייך צינור הנתונים הזה. חובה אם האפשרות Commit Offsets to Kafka (שמירת היסטוגרמת המיקומים ב-Kafka) מופעלת. ברירת המחדל היא ריקה.
• ‫kafkaReadOffset: נקודת ההתחלה לקריאת הודעות כשאין היסטים מחויבים. ההתחלה המוקדמת ביותר היא מההתחלה, וההתחלה המאוחרת ביותר היא מההודעה החדשה ביותר. ברירת המחדל היא: latest.
• ‫kafkaReadUsernameSecretId: מזהה הסוד ב-Google Cloud Secret Manager שמכיל את שם המשתמש ב-Kafka שבו יש להשתמש עם אימות SASL_PLAIN. לדוגמה, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. ברירת המחדל היא ריקה.
• ‫kafkaReadPasswordSecretId: מזהה הסוד ב-Google Cloud Secret Manager שמכיל את הסיסמה ל-Kafka שבה יש להשתמש לאימות SASL_PLAIN. לדוגמה, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. ברירת המחדל היא ריקה.
• ‫kafkaReadKeystoreLocation: הנתיב ב-Google Cloud Storage לקובץ Java KeyStore ‏ (JKS) שמכיל את אישור ה-TLS והמפתח הפרטי שבהם יש להשתמש בעת אימות מול אשכול Kafka. לדוגמה, gs://your-bucket/keystore.jks.
• ‫kafkaReadTruststoreLocation: הנתיב ב-Google Cloud Storage לקובץ Java TrustStore‏ (JKS) שמכיל את האישורים המהימנים שמשמשים לאימות הזהות של ברוקר Kafka.
• ‫kafkaReadTruststorePasswordSecretId: מזהה הסוד ב-Google Cloud Secret Manager שמכיל את הסיסמה לגישה לקובץ Java TrustStore ‏ (JKS) לאימות TLS של Kafka. לדוגמה, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• ‫kafkaReadKeystorePasswordSecretId: מזהה הסוד ב-Google Cloud Secret Manager שמכיל את הסיסמה לגישה לקובץ Java KeyStore ‏ (JKS) לאימות Kafka TLS. לדוגמה, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• ‫kafkaReadKeyPasswordSecretId: מזהה הסוד ב-Google Cloud Secret Manager שמכיל את הסיסמה לגישה למפתח הפרטי בקובץ Java KeyStore ‏ (JKS) לאימות TLS של Kafka. לדוגמה, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• ‫kafkaReadSaslScramUsernameSecretId: המזהה הסודי של Google Cloud Secret Manager שמכיל את שם המשתמש של Kafka לשימוש באימות SASL_SCRAM. לדוגמה, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• ‫kafkaReadSaslScramPasswordSecretId: מזהה הסוד ב-Google Cloud Secret Manager שמכיל את הסיסמה ל-Kafka לשימוש באימות SASL_SCRAM. לדוגמה, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• ‫kafkaReadSaslScramTruststoreLocation: הנתיב ב-Google Cloud Storage לקובץ Java TrustStore‏ (JKS) שמכיל את האישורים המהימנים שמשמשים לאימות הזהות של ברוקר Kafka.
• ‫kafkaReadSaslScramTruststorePasswordSecretId: מזהה הסוד ב-Google Cloud Secret Manager שמכיל את הסיסמה לגישה לקובץ Java TrustStore ‏ (JKS) לאימות Kafka SASL_SCRAM. לדוגמה, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• ‫schemaFormat: פורמט הסכימה של Kafka. אפשר לספק את הערך כ-SINGLE_SCHEMA_FILE או כ-SCHEMA_REGISTRY. אם מציינים את SINGLE_SCHEMA_FILE, צריך להשתמש בסכימה שמוזכרת בקובץ הסכימה של avro לכל ההודעות. אם מציינים את SCHEMA_REGISTRY, ההודעות יכולות להיות עם סכימה אחת או עם כמה סכימות. ברירת המחדל היא: SINGLE_SCHEMA_FILE.
• ‫confluentAvroSchemaPath: הנתיב ב-Google Cloud Storage לקובץ סכמת Avro יחיד שמשמש לפענוח כל ההודעות בנושא. ברירת המחדל היא ריקה.
• ‫schemaRegistryConnectionUrl: כתובת ה-URL של מופע Confluent Schema Registry שמשמש לניהול סכימות Avro לפענוח הודעות. ברירת המחדל היא ריקה.
• ‫binaryAvroSchemaPath: הנתיב ב-Google Cloud Storage לקובץ סכימת Avro שמשמש לפענוח הודעות Avro בקידוד בינארי. ברירת המחדל היא ריקה.
• ‫schemaRegistryAuthenticationMode: מצב האימות לשימוש עם Schema Registry. משתמשים ב-NONE ללא אימות, ב-TLS לאימות מבוסס-אישור וב-OAUTH לאימות OAuth2. ‫APPLICATION_DEFAULT_CREDENTIALS צריך לשמש רק לשירות מנוהל של Google Cloud ל-Apache Kafka Schema Registry, והוא מאפשר אימות באמצעות פרטי כניסה שמוגדרים כברירת מחדל באפליקציה.
• ‫schemaRegistryTruststoreLocation: המיקום של אישור ה-SSL שבו מאוחסן מאגר האישורים לאימות ב-Schema Registry. לדוגמה, /your-bucket/truststore.jks.
• ‫schemaRegistryTruststorePasswordSecretId: SecretId במנהל הסודות שבו מאוחסנת הסיסמה לגישה לסוד ב-truststore. לדוגמה, projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
• ‫schemaRegistryKeystoreLocation: מיקום ה-Keystore שמכיל את אישור ה-SSL והמפתח הפרטי. לדוגמה, /your-bucket/keystore.jks.
• ‫schemaRegistryKeystorePasswordSecretId: SecretId ב-Secret Manager שבו הסיסמה לגישה לקובץ של מאגר המפתחות. לדוגמה, projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
• ‫schemaRegistryKeyPasswordSecretId: SecretId של הסיסמה שנדרשת לגישה למפתח הפרטי של הלקוח שמאוחסן במאגר המפתחות. לדוגמה, projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
• ‫schemaRegistryOauthClientId: מזהה הלקוח שמשמש לאימות לקוח Schema Registry במצב OAUTH. נדרש לפורמט ההודעה AVRO_CONFLUENT_WIRE_FORMAT.
• ‫schemaRegistryOauthClientSecretId: מזהה הסוד ב-Google Cloud Secret Manager שמכיל את סוד הלקוח שמשמש לאימות לקוח Schema Registry במצב OAUTH. נדרש לפורמט ההודעה AVRO_CONFLUENT_WIRE_FORMAT. לדוגמה, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• ‫schemaRegistryOauthScope: היקף אסימון הגישה שמשמש לאימות לקוח Schema Registry במצב OAUTH. השדה הזה הוא אופציונלי, כי אפשר לשלוח את הבקשה בלי להעביר פרמטר היקף. לדוגמה, openid.
• ‫schemaRegistryOauthTokenEndpointUrl: כתובת ה-URL מבוססת ה-HTTP(S) של ספק הזהויות OAuth/OIDC שמשמש לאימות לקוח Schema Registry במצב OAUTH. נדרש לפורמט ההודעה AVRO_CONFLUENT_WIRE_FORMAT.
• ‫outputDeadletterTable: שם הטבלה ב-BigQuery שכולל את שם הפרויקט, מערך הנתונים והטבלה, עבור הודעות שנכשלו. הודעות שלא הגיעו לטבלת הפלט מסיבות שונות (למשל, סכימה לא תואמת, קובץ JSON לא תקין) נכתבות בטבלה הזו. הטבלה נוצרת על ידי התבנית. לדוגמה, your-project-id:your-dataset.your-table-name.

הרצת התבנית

המסוף

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

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

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

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/VERSION/flex/Kafka_to_Gcs_Flex \
    --parameters \
readBootstrapServerAndTopic=BOOTSTRAP_SERVER_AND_TOPIC,\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
messageFormat=JSON,\
outputDirectory=gs://STORAGE_BUCKET_NAME,\
useBigQueryDLQ=false
  

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

• ‫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/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫BOOTSTRAP_SERVER_AND_TOPIC: כתובת שרת ה-bootstrap והנושא של Apache Kafka

  הפורמט של כתובת שרת האתחול והנושא תלוי בסוג האשכול:

  • קלאסטר שירות מנוהל ל-Apache Kafka: projects/PROJECT_ID/locations/REGION_NAME/clusters/CLUSTER_NAME/topics/TOPIC_NAME
  • אשכול Kafka חיצוני: BOOTSTRAP_SERVER_ADDRESS;TOPIC_NAME
• ‫STORAGE_BUCKET_NAME: הקטגוריה של Cloud Storage שבה נכתב הפלט

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": {
          "readBootstrapServerAndTopic": "BOOTSTRAP_SERVER_AND_TOPIC",
          "kafkaReadAuthenticationMode": "APPLICATION_DEFAULT_CREDENTIALS",
          "messageFormat": "JSON",
          "outputDirectory": "gs://STORAGE_BUCKET_NAME",
          "useBigQueryDLQ": "false"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/Kafka_to_Gcs_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/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫BOOTSTRAP_SERVER_AND_TOPIC: כתובת שרת ה-bootstrap והנושא של Apache Kafka

  הפורמט של כתובת שרת האתחול והנושא תלוי בסוג האשכול:

  • קלאסטר שירות מנוהל ל-Apache Kafka: projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_NAME/topics/TOPIC_NAME
  • אשכול Kafka חיצוני: BOOTSTRAP_SERVER_ADDRESS;TOPIC_NAME
• ‫STORAGE_BUCKET_NAME: הקטגוריה של Cloud Storage שבה נכתב הפלט

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

Java

/*
 * Copyright (C) 2024 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.templates;

import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.v2.kafka.dlq.BigQueryDeadLetterQueue;
import com.google.cloud.teleport.v2.kafka.dlq.BigQueryDeadLetterQueueOptions;
import com.google.cloud.teleport.v2.kafka.options.KafkaReadOptions;
import com.google.cloud.teleport.v2.kafka.options.SchemaRegistryOptions;
import com.google.cloud.teleport.v2.kafka.transforms.KafkaTransform;
import com.google.cloud.teleport.v2.kafka.utils.KafkaConfig;
import com.google.cloud.teleport.v2.kafka.utils.KafkaTopicUtils;
import com.google.cloud.teleport.v2.transforms.WriteTransform;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.io.kafka.KafkaRecord;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.errorhandling.BadRecord;
import org.apache.beam.sdk.transforms.errorhandling.BadRecordRouter;
import org.apache.beam.sdk.transforms.errorhandling.ErrorHandler;
import org.apache.beam.sdk.values.PCollection;

@Template(
    name = "Kafka_to_Gcs_Flex",
    category = TemplateCategory.STREAMING,
    displayName = "Kafka to Cloud Storage",
    description =
        "A streaming pipeline which ingests data from Kafka and writes to a pre-existing Cloud"
            + " Storage bucket with a variety of file types.",
    optionsClass = KafkaToGcsFlex.KafkaToGcsOptions.class,
    flexContainerName = "kafka-to-gcs",
    contactInformation = "https://cloud.google.com/support",
    requirements = {"The output Google Cloud Storage directory must exist."})
public class KafkaToGcsFlex {

  public interface KafkaToGcsOptions
      extends PipelineOptions,
          DataflowPipelineOptions,
          KafkaReadOptions,
          SchemaRegistryOptions,
          BigQueryDeadLetterQueueOptions {

    // This is a duplicate option that already exist in KafkaReadOptions but keeping it here
    // so the KafkaTopic appears above the authentication enum on the Templates UI.
    @TemplateParameter.KafkaReadTopic(
        order = 1,
        name = "readBootstrapServerAndTopic",
        groupName = "Source",
        description = "Source Kafka Topic",
        helpText = "Kafka Topic to read the input from.")
    String getReadBootstrapServerAndTopic();

    void setReadBootstrapServerAndTopic(String value);

    @TemplateParameter.Duration(
        order = 20,
        optional = true,
        groupName = "Destination",
        description = "Window duration",
        helpText =
            "The window duration/size in which data will be written to Cloud Storage. Allowed formats are: Ns (for "
                + "seconds, example: 5s), Nm (for minutes, example: 12m), Nh (for hours, example: 2h).",
        example = "5m")
    @Default.String("5m")
    String getWindowDuration();

    void setWindowDuration(String windowDuration);

    @TemplateParameter.GcsWriteFolder(
        order = 21,
        groupName = "Destination",
        description = "Output file directory in Cloud Storage",
        helpText = "The path and filename prefix for writing output files. Must end with a slash.",
        example = "gs://your-bucket/your-path/")
    String getOutputDirectory();

    void setOutputDirectory(String outputDirectory);

    @TemplateParameter.Text(
        order = 22,
        optional = true,
        groupName = "Destination",
        description = "Output filename prefix of the files to write",
        helpText = "The prefix to place on each windowed file.",
        example = "output-")
    @Default.String("output")
    String getOutputFilenamePrefix();

    void setOutputFilenamePrefix(String outputFilenamePrefix);

    @TemplateParameter.Integer(
        order = 23,
        optional = true,
        description = "Maximum output shards",
        groupName = "Destination",
        helpText =
            "The maximum number of output shards produced when writing. A higher number of "
                + "shards means higher throughput for writing to Cloud Storage, but potentially higher "
                + "data aggregation cost across shards when processing output Cloud Storage files. "
                + "Default value is decided by Dataflow.")
    @Default.Integer(0)
    Integer getNumShards();

    void setNumShards(Integer numShards);
  }

  public static PipelineResult run(KafkaToGcsOptions options) throws Exception {
    // Create the Pipeline
    Pipeline pipeline = Pipeline.create(options);
    String bootstrapServes;
    List<String> topicsList;
    if (options.getReadBootstrapServerAndTopic() != null) {
      List<String> bootstrapServerAndTopicList =
          KafkaTopicUtils.getBootstrapServerAndTopic(
              options.getReadBootstrapServerAndTopic(), options.getProject());
      topicsList = List.of(bootstrapServerAndTopicList.get(1));
      bootstrapServes = bootstrapServerAndTopicList.get(0);
    } else {
      throw new IllegalArgumentException(
          "Please provide a valid bootstrap server which matches `[,:a-zA-Z0-9._-]+` and a topic which matches `[,a-zA-Z0-9._-]+`");
    }

    options.setStreaming(true);

    Map<String, Object> kafkaConfig = new HashMap<>(KafkaConfig.fromReadOptions(options));

    // Configure dead letter queue params
    ErrorHandler<BadRecord, ?> errorHandler = new ErrorHandler.DefaultErrorHandler<>();
    // Throwing Router throws the error instead of sending it to the DLQ. This will be the case
    // when no DLQ is configured and the pipeline will retry the failed error.
    BadRecordRouter badRecordRouter = BadRecordRouter.THROWING_ROUTER;

    if (options.getUseBigQueryDLQ()) {
      if (options.getOutputDeadletterTable() == null
          || options.getOutputDeadletterTable().isBlank()) {
        throw new IllegalArgumentException(
            "Please provide a Fully Qualified BigQuery table name when BigQuery Dead Letter"
                + "Queue is enabled");
      }
      badRecordRouter = BadRecordRouter.RECORDING_ROUTER;
      errorHandler =
          pipeline.registerBadRecordErrorHandler(
              BigQueryDeadLetterQueue.newBuilder()
                  .setTableName(options.getOutputDeadletterTable())
                  .build());
    }

    PCollection<KafkaRecord<byte[], byte[]>> kafkaRecord;
    // Step 1: Read from Kafka as bytes.
    KafkaIO.Read<byte[], byte[]> kafkaTransform =
        KafkaTransform.readBytesFromKafka(
            bootstrapServes, topicsList, kafkaConfig, options.getEnableCommitOffsets());
    kafkaRecord = pipeline.apply(kafkaTransform);

    kafkaRecord.apply(
        WriteTransform.newBuilder()
            .setOptions(options)
            .setBadRecordErrorHandler(errorHandler)
            .setBadRecordRouter(badRecordRouter)
            .build());
    if (options.getUseBigQueryDLQ()) {
      errorHandler.close();
    }
    return pipeline.run();
  }

  public static void main(String[] args) throws Exception {
    KafkaToGcsOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(KafkaToGcsOptions.class);

    run(options);
  }
}

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

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