תבנית Apache Kafka ל-BigQuery

תבנית Apache Kafka ל-BigQuery היא פייפליין להזרמת נתונים שקולט נתוני טקסט מאשכולות של שירות מנוהל של Google Cloud ל-Apache Kafka, ואז מוציא את הרשומות שמתקבלות לטבלאות ב-BigQuery. כל השגיאות שמתרחשות במהלך הוספת הנתונים לטבלת הפלט מתווספות לטבלת שגיאות נפרדת ב-BigQuery.

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

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

• שרת הברוקר של Apache Kafka צריך לפעול ולהיות נגיש ממכונות העובד של Dataflow.
• נושאי Apache Kafka חייבים להתקיים.
• צריך להפעיל את ממשקי ה-API של Dataflow,‏ BigQuery ו-Cloud Storage. אם נדרש אימות, צריך גם להפעיל את Secret Manager API.
• יוצרים מערך נתונים וטבלה ב-BigQuery עם הסכימה המתאימה לנושא הקלט של Kafka. אם משתמשים בכמה סכימות באותו נושא ורוצים לכתוב לכמה טבלאות, לא צריך ליצור את הטבלה לפני שמגדירים את צינור הנתונים.
• כשמפעילים את תור ההודעות שלא עברו עיבוד (dead-letter) של התבנית, צריך ליצור טבלה ריקה בלי סכימה לתור ההודעות שלא עברו עיבוד.
• אם מתחברים לאשכול של שירות מנוהל ל-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.

אימות

תבנית Apache Kafka ל-BigQuery תומכת באימות SASL/PLAIN לשרתי Kafka.

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

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

• ‫readBootstrapServerAndTopic: נושא Kafka לקריאת הקלט ממנו.
• ‫writeMode: כתיבת רשומות לטבלה אחת או לכמה טבלאות (על סמך הסכימה). מצב DYNAMIC_TABLE_NAMES נתמך רק בAVRO_CONFLUENT_WIRE_FORMAT פורמט הודעת מקור ובSCHEMA_REGISTRY מקור סכימה. שם טבלת היעד נוצר באופן אוטומטי על סמך שם סכימת Avro של כל הודעה. יכולה להיות סכימה אחת (שיוצרת טבלה אחת) או כמה סכימות (שיוצרות כמה טבלאות). במצב SINGLE_TABLE_NAME, המערכת כותבת לטבלה אחת (סכימה אחת) שהמשתמש מציין. ברירת המחדל היא SINGLE_TABLE_NAME.
• ‫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.

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

• ‫outputTableSpec: מיקום הטבלה ב-BigQuery שאליה ייכתב הפלט. השם צריך להיות בפורמט <project>:<dataset>.<table_name>. הסכימה של הטבלה צריכה להתאים לאובייקטים של הקלט.
• ‫persistKafkaKey: אם הערך הוא true, הצינור ישמור את מפתח ההודעה של Kafka בטבלה ב-BigQuery, בשדה _key מסוג BYTES. ברירת המחדל היא false (המערכת מתעלמת מהמפתח).
• ‫outputProject: פרויקט הפלט ב-BigQuery שבו נמצא מערך הנתונים. הטבלאות ייווצרו באופן דינמי במערך הנתונים. ברירת המחדל היא ריק.
• ‫outputDataset: מערך נתוני הפלט של BigQuery שאליו ייכתב הפלט. הטבלאות ייווצרו באופן דינמי במערך הנתונים. אם הטבלאות נוצרות מראש, השמות שלהן צריכים להיות בהתאם למוסכמות למתן שמות שצוינו. השם צריך להיות bqTableNamePrefix + Avro Schema FullName , כל מילה תופרד באמצעות מקף -. ברירת המחדל היא ריק.
• ‫bqTableNamePrefix: קידומת לשם שתשמש ליצירת טבלאות פלט ב-BigQuery. ההגדרה הזו רלוונטית רק כשמשתמשים במאגר סכימות. ברירת המחדל היא ריק.
• ‫createDisposition: BigQuery CreateDisposition. לדוגמה: CREATE_IF_NEEDED, ‏ CREATE_NEVER. ברירת המחדל היא: CREATE_IF_NEEDED.
• ‫writeDisposition: BigQuery WriteDisposition. לדוגמה: WRITE_APPEND, ‏ WRITE_EMPTY או WRITE_TRUNCATE. ברירת המחדל היא: WRITE_APPEND.
• ‫useAutoSharding: אם הערך הוא true, הצינור משתמש בפיצול אוטומטי של נתונים (auto-sharding) כשמתבצעת כתיבה ל-BigQuery. ערך ברירת המחדל הוא true.
• ‫numStorageWriteApiStreams: מציין את מספר זרמי הכתיבה. חובה להגדיר את הפרמטר הזה. ערך ברירת המחדל הוא 0.
• ‫storageWriteApiTriggeringFrequencySec: הפרמטר הזה מציין את תדירות ההפעלה בשניות, וחובה להגדיר אותו. ברירת המחדל היא 5 שניות.
• ‫useStorageWriteApiAtLeastOnce: הפרמטר הזה תקף רק אם האפשרות 'שימוש ב-BigQuery Storage Write API' מופעלת. אם האפשרות הזו מופעלת, נעשה שימוש בסמנטיקה של מסירה לפחות פעם אחת עבור Storage Write API. אחרת, נעשה שימוש בסמנטיקה של מסירה חד-פעמית. ברירת המחדל היא: false.
• ‫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) לאימות Kafka TLS. לדוגמה, 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, והוא מאפשר אימות באמצעות Application Default Credentials.
• ‫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.
• ‫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: מציין באיזו תדירות לטעון מחדש את ה-UDF, בדקות. אם הערך גדול מ-0, מערכת Dataflow בודקת מעת לעת את קובץ ה-UDF ב-Cloud Storage, ומטעינה מחדש את ה-UDF אם הקובץ משתנה. הפרמטר הזה מאפשר לכם לעדכן את הפונקציה המוגדרת על ידי המשתמש בזמן שהצינור פועל, בלי שתצטרכו להפעיל מחדש את העבודה. אם הערך הוא 0, טעינה מחדש של פונקציות UDF מושבתת. ערך ברירת המחדל הוא 0.

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

אפשר גם להרחיב את התבנית הזו על ידי כתיבת פונקציה בהגדרת המשתמש (UDF). התבנית קוראת ל-UDF לכל רכיב קלט. מטענים ייעודיים של רכיבים עוברים סריאליזציה כמחרוזות JSON. למידע נוסף, ראו יצירת פונקציות מוגדרות על ידי המשתמש לתבניות Dataflow.

התבנית תומכת בפונקציות UDF רק בהודעות Kafka בפורמט JSON. אם ההודעות ב-Kafka הן בפורמט Avro, הפונקציה המוגדרת על ידי המשתמש לא מופעלת.

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

המאפיינים של פונקציית UDF:

• קלט: ערך הרשומה ב-Kafka, שעבר סריאליזציה כמחרוזת JSON
• פלט: מחרוזת JSON שתואמת לסכימה של טבלת היעד ב-BigQuery

הרצת התבנית

המסוף

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

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

4. בתפריט הנפתח Dataflow template (תבנית Dataflow), בוחרים את התבנית Kafka to BigQuery (מ-Kafka ל-BigQuery).
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-REGION_NAME/VERSION/flex/ \
    --parameters \
readBootstrapServerAndTopic=BOOTSTRAP_SERVER_AND_TOPIC,\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
messageFormat=JSON,\
writeMode=SINGLE_TABLE_NAME,\
outputTableSpec=PROJECT_ID:DATASET_NAME.TABLE_NAME,\
useBigQueryDLQ=true,\
outputDeadletterTable=PROJECT_ID:DATASET_NAME.ERROR_TABLE_NAME
  

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

• ‫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/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫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
• ‫DATASET_NAME: השם של מערך הנתונים ב-BigQuery
• ‫TABLE_NAME: השם של טבלת הפלט ב-BigQuery
• ‫ERROR_TABLE_NAME: השם של הטבלה ב-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": {
          "readBootstrapServerAndTopic": "BOOTSTRAP_SERVER_AND_TOPIC",
          "kafkaReadAuthenticationMode": "APPLICATION_DEFAULT_CREDENTIALS",
          "messageFormat": "JSON",
          "writeMode": "SINGLE_TABLE_NAME",
          "outputTableSpec": "PROJECT_ID:DATASET_NAME.TABLE_NAME",
          "useBigQueryDLQ": "true",
          "outputDeadletterTable": "PROJECT_ID:DATASET_NAME.ERROR_TABLE_NAME"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/",
   }
}
  

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

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

מידע נוסף זמין במאמר כתיבת נתונים מ-Kafka ל-BigQuery באמצעות Dataflow.

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

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.api.services.bigquery.model.TableRow;
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.kafka.dlq.BigQueryDeadLetterQueue;
import com.google.cloud.teleport.v2.kafka.transforms.AvroDynamicTransform;
import com.google.cloud.teleport.v2.kafka.transforms.AvroTransform;
import com.google.cloud.teleport.v2.kafka.transforms.KafkaRecordErrorConverters.WriteKafkaRecordMessageErrors;
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.kafka.values.KafkaTemplateParameters.MessageFormatConstants;
import com.google.cloud.teleport.v2.kafka.values.KafkaTemplateParameters.SchemaFormat;
import com.google.cloud.teleport.v2.options.KafkaToBigQueryFlexOptions;
import com.google.cloud.teleport.v2.transforms.BigQueryWriteUtils;
import com.google.cloud.teleport.v2.transforms.ErrorConverters;
import com.google.cloud.teleport.v2.transforms.StringMessageToTableRow;
import com.google.cloud.teleport.v2.utils.BigQueryIOUtils;
import com.google.cloud.teleport.v2.utils.MetadataValidator;
import com.google.cloud.teleport.v2.utils.SchemaUtils;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import com.google.common.base.Strings;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.ByteArrayCoder;
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.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryInsertError;
import org.apache.beam.sdk.io.gcp.bigquery.InsertRetryPolicy;
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult;
import org.apache.beam.sdk.io.kafka.KafkaRecord;
import org.apache.beam.sdk.io.kafka.KafkaRecordCoder;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
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;
import org.apache.beam.sdk.values.PCollectionList;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.commons.lang3.ObjectUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link KafkaToBigQueryFlex} pipeline is a streaming pipeline which ingests text data from
 * Kafka, and outputs the resulting records to BigQuery. Any errors which occur in the
 * transformation of the data, or inserting into the output table will be inserted into a separate
 * errors table in BigQuery. Both output and error tables are specified by the user as parameters.
 *
 * <p><b>Pipeline Requirements</b>
 *
 * <ul>
 *   <li>The Kafka topic exists and the message is encoded in a valid JSON format.
 *   <li>The BigQuery output table exists.
 *   <li>The Kafka brokers are reachable from the Dataflow worker machines.
 * </ul>
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/kafka-to-bigquery/README_Kafka_to_BigQuery.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Kafka_to_BigQuery_Flex",
    category = TemplateCategory.STREAMING,
    displayName = "Kafka to BigQuery",
    description =
        "The Apache Kafka to BigQuery template is a streaming pipeline which ingests text data from Apache Kafka, and outputs the resulting records to BigQuery. "
            + "Any errors which occur in the transformation of the data, or inserting into the output table are inserted into a separate errors table in BigQuery. "
            + "For any errors which occur in the transformation of the data, the original records can be inserted into a separate Kafka topic. The template supports "
            + "reading a Kafka topic which contains single/multiple schema(s). It can write to a single or multiple BigQuery tables, depending on the schema of records. ",
    optionsClass = KafkaToBigQueryFlexOptions.class,
    flexContainerName = "kafka-to-bigquery",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/kafka-to-bigquery",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "The output BigQuery table must exist.",
      "The Apache Kafka broker server must be running and be reachable from the Dataflow worker machines.",
      "The Apache Kafka topics must exist and the messages must be encoded in a valid JSON format."
    },
    skipOptions = {"useStorageWriteApi"})
public class KafkaToBigQueryFlex {

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

  /** The tag for the main output of the json transformation. */
  public static final TupleTag<TableRow> TRANSFORM_OUT = new TupleTag<>() {};

  /** The tag for the dead-letter output of the json to table row transform. */
  public static final TupleTag<FailsafeElement<KafkaRecord<String, String>, String>>
      TRANSFORM_DEADLETTER_OUT = new TupleTag<>() {};

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

  private static ErrorHandler<BadRecord, ?> errorHandler = new ErrorHandler.DefaultErrorHandler<>();
  private static BadRecordRouter badRecordRouter = BadRecordRouter.THROWING_ROUTER;

  /**
   * The main entry-point for pipeline execution. This method will start the pipeline but will not
   * wait for its execution to finish. If blocking execution is required, use the {@link
   * KafkaToBigQueryFlex#run(KafkaToBigQueryFlexOptions)} method to start the pipeline and invoke
   * {@code result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line args passed by the executor.
   */
  public static void main(String[] args) throws Exception {
    UncaughtExceptionLogger.register();

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

    run(options);
  }

  public static Boolean useErrorHandler(KafkaToBigQueryFlexOptions options) {
    return options.getUseBigQueryDLQ();
  }

  public static WriteResult processKafkaRecords(
      PCollection<KafkaRecord<byte[], byte[]>> kafkaRecords, KafkaToBigQueryFlexOptions options) {

    // Validate the pipeline options for MessageFormat and SchemaFormat.
    if (options.getMessageFormat().equals(MessageFormatConstants.AVRO_BINARY_ENCODING)
        && (options.getBinaryAvroSchemaPath() != null
            && options.getBinaryAvroSchemaPath().isBlank())) {
      throw new IllegalArgumentException(
          "Binary Avro Schema Path cannot be empty for AVRO_BINARY_ENCODING.");
    }

    if (options.getMessageFormat().equals(MessageFormatConstants.AVRO_CONFLUENT_WIRE_FORMAT)) {

      if ((options.getSchemaRegistryConnectionUrl() != null
              && options.getSchemaRegistryConnectionUrl().isBlank())
          && (options.getConfluentAvroSchemaPath() != null
              && options.getConfluentAvroSchemaPath().isBlank())) {
        throw new IllegalArgumentException(
            "Either Schema Registry Connection URL or Confluent Avro Schema Path must be provided for AVRO_CONFLUENT_WIRE_FORMAT.");
      }

      if (options.getSchemaFormat().equals(SchemaFormat.SINGLE_SCHEMA_FILE)) {
        if (!options.getConfluentAvroSchemaPath().isBlank()
            && (options.getOutputTableSpec() != null && options.getOutputTableSpec().isBlank())) {
          throw new IllegalArgumentException(
              "The outputTableSpec parameter is required when using the SINGLE_SCHEMA_FILE schema format.");
        }
      } else if (options.getSchemaFormat().equals(SchemaFormat.SCHEMA_REGISTRY)) {
        if (options.getSchemaRegistryConnectionUrl() != null
            && (options.getOutputDataset() != null && options.getOutputDataset().isBlank())) {
          throw new IllegalArgumentException(
              "An output BigQuery dataset is required. It will be used to create tables per schema.");
        }
      } else {
        throw new IllegalArgumentException(
            "Unsupported schemaFormat parameter value: " + options.getSchemaFormat());
      }
    }

    if (options.getMessageFormat().equals(MessageFormatConstants.AVRO_BINARY_ENCODING)
        && (!options.getBinaryAvroSchemaPath().isBlank())) {
      return handleAvroBinaryEncoding(kafkaRecords, options);
    } else if (options.getMessageFormat().equals(MessageFormatConstants.AVRO_CONFLUENT_WIRE_FORMAT)
        && (!options.getSchemaRegistryConnectionUrl().isBlank()
            || !options.getConfluentAvroSchemaPath().isBlank())) {
      return handleAvroConfluentWireFormat(kafkaRecords, options);
    } else {
      throw new IllegalArgumentException(
          "Message format " + options.getMessageFormat() + " is unsupported.");
    }
  }

  /**
   * Runs the pipeline to completion with the specified options. This method does not wait until the
   * pipeline is finished before returning. Invoke {@code result.waitUntilFinish()} on the result
   * object to block until the pipeline is finished running if blocking programmatic execution is
   * required.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  public static PipelineResult run(KafkaToBigQueryFlexOptions options) throws Exception {

    // Enable Streaming Engine
    options.setEnableStreamingEngine(true);

    List<String> dataflowServiceOptions = options.getDataflowServiceOptions();
    if (dataflowServiceOptions == null) {
      dataflowServiceOptions = new ArrayList<>();
    }
    dataflowServiceOptions.add("enable_streaming_engine_resource_based_billing");
    options.setDataflowServiceOptions(dataflowServiceOptions);

    // Validate BQ STORAGE_WRITE_API options
    options.setUseStorageWriteApi(true);
    if (options.getStorageWriteApiTriggeringFrequencySec() == null) {
      options.setStorageWriteApiTriggeringFrequencySec(5);
    }
    BigQueryIOUtils.validateBQStorageApiOptionsStreaming(options);
    MetadataValidator.validate(options);

    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    List<String> topicsList;
    String bootstrapServers;
    if (options.getReadBootstrapServerAndTopic() != null) {
      List<String> bootstrapServerAndTopicList =
          KafkaTopicUtils.getBootstrapServerAndTopic(
              options.getReadBootstrapServerAndTopic(), options.getProject());
      topicsList = List.of(bootstrapServerAndTopicList.get(1));
      bootstrapServers = 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._-]+`");
    }

    // Configure error handler for Dead letter queue
    if (options.getUseBigQueryDLQ()) {
      if (options.getOutputDeadletterTable() == null
          || options.getOutputDeadletterTable().isBlank()) {
        throw new IllegalArgumentException(
            "Please provide a valid BigQuery full qualified table name when using BigQuery"
                + "Dead letter queue");
      }
      badRecordRouter = BadRecordRouter.RECORDING_ROUTER;
      errorHandler =
          pipeline.registerBadRecordErrorHandler(
              BigQueryDeadLetterQueue.newBuilder()
                  .setTableName(options.getOutputDeadletterTable())
                  .build());
    }

    // Get the Kafka config
    Map<String, Object> kafkaConfig = KafkaConfig.fromReadOptions(options);

    /*
     * Steps:
     *  1) Read messages in from Kafka
     *  2) Transform the messages into TableRows
     *     - Transform message payload via UDF
     *     - Convert UDF result to TableRow objects
     *  3) Write successful records out to BigQuery
     *  4) Write failed records out to BigQuery
     */

    if (options.getMessageFormat() == null
        || options.getMessageFormat().equals(MessageFormatConstants.JSON)) {

      pipeline = runJsonPipeline(pipeline, options, topicsList, bootstrapServers, kafkaConfig);

    } else if (options.getMessageFormat().equals(MessageFormatConstants.AVRO_CONFLUENT_WIRE_FORMAT)
        || options.getMessageFormat().equals(MessageFormatConstants.AVRO_BINARY_ENCODING)) {
      pipeline = runAvroPipeline(pipeline, options, topicsList, bootstrapServers, kafkaConfig);

    } else {
      throw new IllegalArgumentException("Invalid format specified: " + options.getMessageFormat());
    }
    if (useErrorHandler(options)) {
      errorHandler.close();
    }
    return pipeline.run();
  }

  private static WriteResult handleAvroBinaryEncoding(
      PCollection<KafkaRecord<byte[], byte[]>> kafkaRecords, KafkaToBigQueryFlexOptions options) {
    WriteResult writeResult;
    BigQueryWriteUtils.BigQueryWrite bigQueryWrite;
    if (useErrorHandler(options)) {
      // BigQueryIO sets the BadRecordRouter to RecordingRouter even when the errorHandler is
      // DefaultErrorHandler(which is a no op). In this case, when the BadRecordRouter is
      // ThrowingRouter,
      // don't pass errorHandler to BigQueryIO.
      bigQueryWrite =
          BigQueryWriteUtils.BigQueryWrite.of(
              SchemaUtils.getAvroSchema(options.getBinaryAvroSchemaPath()),
              options.getOutputTableSpec(),
              options.getWriteDisposition(),
              options.getCreateDisposition(),
              options.getNumStorageWriteApiStreams(),
              options.getStorageWriteApiTriggeringFrequencySec(),
              options.getPersistKafkaKey(),
              options.getUseAutoSharding(),
              errorHandler);
    } else {
      bigQueryWrite =
          BigQueryWriteUtils.BigQueryWrite.of(
              SchemaUtils.getAvroSchema(options.getBinaryAvroSchemaPath()),
              options.getOutputTableSpec(),
              options.getWriteDisposition(),
              options.getCreateDisposition(),
              options.getNumStorageWriteApiStreams(),
              options.getStorageWriteApiTriggeringFrequencySec(),
              options.getPersistKafkaKey(),
              options.getUseAutoSharding());
    }
    writeResult =
        kafkaRecords
            .apply(
                AvroTransform.of(
                    options.getMessageFormat(),
                    options.getBinaryAvroSchemaPath(),
                    errorHandler,
                    badRecordRouter))
            .apply(bigQueryWrite);
    return writeResult;
  }

  private static WriteResult handleAvroConfluentWireFormat(
      PCollection<KafkaRecord<byte[], byte[]>> kafkaRecords, KafkaToBigQueryFlexOptions options) {
    if (options.getSchemaFormat().equals(SchemaFormat.SINGLE_SCHEMA_FILE)) {
      return handleSingleSchemaFileFormat(kafkaRecords, options);
    } else if (options.getSchemaFormat().equals(SchemaFormat.SCHEMA_REGISTRY)) {
      return handleSchemaRegistryFormat(kafkaRecords, options);
    } else {
      throw new IllegalArgumentException(
          "Message format " + options.getSchemaFormat() + " is unsupported.");
    }
  }

  private static WriteResult handleSingleSchemaFileFormat(
      PCollection<KafkaRecord<byte[], byte[]>> kafkaRecords, KafkaToBigQueryFlexOptions options) {
    if (!(options.getConfluentAvroSchemaPath() != null && options.getOutputTableSpec() != null)) {
      // TODO: Add error.
      throw new RuntimeException("");
    }
    WriteResult writeResult;
    BigQueryWriteUtils.BigQueryWrite bigQueryWrite;
    if (useErrorHandler(options)) {
      bigQueryWrite =
          BigQueryWriteUtils.BigQueryWrite.of(
              SchemaUtils.getAvroSchema(options.getConfluentAvroSchemaPath()),
              options.getOutputTableSpec(),
              options.getWriteDisposition(),
              options.getCreateDisposition(),
              options.getNumStorageWriteApiStreams(),
              options.getStorageWriteApiTriggeringFrequencySec(),
              options.getPersistKafkaKey(),
              options.getUseAutoSharding(),
              errorHandler);
    } else {
      bigQueryWrite =
          BigQueryWriteUtils.BigQueryWrite.of(
              SchemaUtils.getAvroSchema(options.getConfluentAvroSchemaPath()),
              options.getOutputTableSpec(),
              options.getWriteDisposition(),
              options.getCreateDisposition(),
              options.getNumStorageWriteApiStreams(),
              options.getStorageWriteApiTriggeringFrequencySec(),
              options.getPersistKafkaKey(),
              options.getUseAutoSharding());
    }
    writeResult =
        kafkaRecords
            .apply(
                AvroTransform.of(
                    options.getMessageFormat(),
                    options.getConfluentAvroSchemaPath(),
                    errorHandler,
                    badRecordRouter))
            .apply(bigQueryWrite);
    return writeResult;
  }

  private static WriteResult handleSchemaRegistryFormat(
      PCollection<KafkaRecord<byte[], byte[]>> kafkaRecords, KafkaToBigQueryFlexOptions options) {
    if (!(options.getSchemaRegistryConnectionUrl() != null && options.getOutputDataset() != null)) {
      throw new RuntimeException(
          "Missing required parameters: Schema Registry URL and/or Output Dataset");
    }
    WriteResult writeResult;
    BigQueryWriteUtils.BigQueryDynamicWrite bigQueryWrite;
    if (useErrorHandler(options)) {
      bigQueryWrite =
          BigQueryWriteUtils.BigQueryDynamicWrite.of(
              options.getOutputProject(),
              options.getOutputDataset(),
              options.getBqTableNamePrefix(),
              options.getWriteDisposition(),
              options.getCreateDisposition(),
              options.getNumStorageWriteApiStreams(),
              options.getStorageWriteApiTriggeringFrequencySec(),
              options.getPersistKafkaKey(),
              options.getUseAutoSharding(),
              errorHandler);
    } else {
      bigQueryWrite =
          BigQueryWriteUtils.BigQueryDynamicWrite.of(
              options.getOutputProject(),
              options.getOutputDataset(),
              options.getBqTableNamePrefix(),
              options.getWriteDisposition(),
              options.getCreateDisposition(),
              options.getNumStorageWriteApiStreams(),
              options.getStorageWriteApiTriggeringFrequencySec(),
              options.getPersistKafkaKey(),
              options.getUseAutoSharding());
    }
    writeResult =
        kafkaRecords
            .apply(
                AvroDynamicTransform.of(
                    options.getSchemaRegistryConnectionUrl(),
                    KafkaConfig.fromSchemaRegistryOptions(options),
                    errorHandler,
                    badRecordRouter))
            .apply(bigQueryWrite);
    return writeResult;
  }

  public static Pipeline runAvroPipeline(
      Pipeline pipeline,
      KafkaToBigQueryFlexOptions options,
      List<String> topicsList,
      String bootstrapServers,
      Map<String, Object> kafkaConfig)
      throws Exception {

    if (options.getMessageFormat().equals(MessageFormatConstants.AVRO_BINARY_ENCODING)
        && options.getBinaryAvroSchemaPath() == null) {
      throw new IllegalArgumentException(
          "Avro schema is needed in order to read non confluent wire format messages.");
    }
    if (options.getMessageFormat().equals(MessageFormatConstants.AVRO_CONFLUENT_WIRE_FORMAT)
        && options.getSchemaRegistryConnectionUrl() == null
        && options.getConfluentAvroSchemaPath() == null) {
      throw new IllegalArgumentException(
          "Schema Registry Connection URL or Avro schema is needed in order to read confluent wire format messages.");
    }
    if (!Strings.isNullOrEmpty(options.getJavascriptTextTransformGcsPath())
        && !Strings.isNullOrEmpty(options.getJavascriptTextTransformFunctionName())) {
      LOG.warn(
          "JavaScript UDF parameters are set while using Avro message format. "
              + "UDFs are supported for JSON format only. No UDF transformation will be applied.");
    }

    PCollection<KafkaRecord<byte[], byte[]>> kafkaRecords;

    kafkaRecords =
        pipeline
            /*
             * Step #1: Read messages in from Kafka and convert to GenericRecords wrap in FailsafeElement
             */
            .apply(
                "ReadBytesFromKafka",
                KafkaTransform.readBytesFromKafka(
                    bootstrapServers, topicsList, kafkaConfig, options.getEnableCommitOffsets()))
            .setCoder(
                KafkaRecordCoder.of(NullableCoder.of(ByteArrayCoder.of()), ByteArrayCoder.of()));

    WriteResult writeResult = processKafkaRecords(kafkaRecords, options);
    return pipeline;
  }

  public static Pipeline runJsonPipeline(
      Pipeline pipeline,
      KafkaToBigQueryFlexOptions options,
      List<String> topicsList,
      String bootstrapServers,
      Map<String, Object> kafkaConfig) {

    // Register the coder for pipeline
    FailsafeElementCoder<KafkaRecord<String, String>, String> coder =
        FailsafeElementCoder.of(
            KafkaRecordCoder.of(
                NullableCoder.of(StringUtf8Coder.of()), NullableCoder.of(StringUtf8Coder.of())),
            NullableCoder.of(StringUtf8Coder.of()));

    CoderRegistry coderRegistry = pipeline.getCoderRegistry();
    coderRegistry.registerCoderForType(coder.getEncodedTypeDescriptor(), coder);

    PCollectionTuple convertedTableRows;
    convertedTableRows =
        pipeline
            /*
             * Step #1: Read messages in from Kafka
             */
            .apply(
                "ReadFromKafka",
                KafkaTransform.readStringFromKafka(
                    bootstrapServers, topicsList, kafkaConfig, options.getEnableCommitOffsets()))

            /*
             * Step #2: Transform the Kafka Messages into TableRows
             */
            .apply(
                "ConvertMessageToTableRow",
                StringMessageToTableRow.newBuilder()
                    .setFileSystemPath(options.getJavascriptTextTransformGcsPath())
                    .setFunctionName(options.getJavascriptTextTransformFunctionName())
                    .setReloadIntervalMinutes(
                        options.getJavascriptTextTransformReloadIntervalMinutes())
                    .setSuccessTag(TRANSFORM_OUT)
                    .setFailureTag(TRANSFORM_DEADLETTER_OUT)
                    .build());
    /*
     * Step #3: Write the successful records out to BigQuery
     */
    WriteResult writeResult =
        convertedTableRows
            .get(TRANSFORM_OUT)
            .apply(
                "WriteSuccessfulRecords",
                BigQueryIO.writeTableRows()
                    .withoutValidation()
                    .withCreateDisposition(CreateDisposition.CREATE_NEVER)
                    .withWriteDisposition(WriteDisposition.WRITE_APPEND)
                    .withExtendedErrorInfo()
                    .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())
                    .to(options.getOutputTableSpec()));

    /*
     * Step 3 Contd.
     * Elements that failed inserts into BigQuery are extracted and converted to FailsafeElement
     */
    PCollection<FailsafeElement<String, String>> failedInserts =
        BigQueryIOUtils.writeResultToBigQueryInsertErrors(writeResult, options)
            .apply(
                "WrapInsertionErrors",
                MapElements.into(FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor())
                    .via(KafkaToBigQueryFlex::wrapBigQueryInsertError))
            .setCoder(FAILSAFE_ELEMENT_CODER);

    if (options.getOutputDeadletterTable() != null) {
      /*
       * Step #4: Write failed records out to BigQuery
       */
      PCollectionList.of(convertedTableRows.get(TRANSFORM_DEADLETTER_OUT))
          .apply("Flatten", Flatten.pCollections())
          .apply(
              "WriteTransformationFailedRecords",
              WriteKafkaRecordMessageErrors.newBuilder()
                  .setErrorRecordsTable(
                      ObjectUtils.firstNonNull(options.getOutputDeadletterTable()))
                  .setErrorRecordsTableSchema(SchemaUtils.DEADLETTER_SCHEMA)
                  .build());
    } else {
      PCollectionList.of(convertedTableRows.get(TRANSFORM_DEADLETTER_OUT))
          .apply("Flatten", Flatten.pCollections())
          .apply("PrintInsertionFailedRecords", ParDo.of(new ThrowErrorFn<>()));
    }

    if (options.getOutputDeadletterTable() != null) {
      /*
       * Step #5: Insert records that failed BigQuery inserts into a dead-letter table.
       */
      failedInserts.apply(
          "WriteInsertionFailedRecords",
          ErrorConverters.WriteStringMessageErrors.newBuilder()
              .setErrorRecordsTable(ObjectUtils.firstNonNull(options.getOutputDeadletterTable()))
              .setErrorRecordsTableSchema(SchemaUtils.DEADLETTER_SCHEMA)
              .build());
    } else {
      failedInserts.apply(
          "PrintInsertionFailedRecords", ParDo.of(new ThrowErrorFn<String, String>()));
    }

    return pipeline;
  }

  /**
   * Method to wrap a {@link BigQueryInsertError} into a {@link FailsafeElement}.
   *
   * @param insertError BigQueryInsert error.
   * @return FailsafeElement object.
   */
  protected static FailsafeElement<String, String> wrapBigQueryInsertError(
      BigQueryInsertError insertError) {

    FailsafeElement<String, String> failsafeElement;
    try {

      failsafeElement =
          FailsafeElement.of(
              insertError.getRow().toPrettyString(), insertError.getRow().toPrettyString());
      failsafeElement.setErrorMessage(insertError.getError().toPrettyString());

    } catch (IOException e) {
      LOG.error("Failed to wrap BigQuery insert error.");
      throw new RuntimeException(e);
    }
    return failsafeElement;
  }

  static class ThrowErrorFn<T, W> extends DoFn<FailsafeElement<T, W>, FailsafeElement<T, W>> {

    @ProcessElement
    public void processElement(ProcessContext context) {
      FailsafeElement<T, W> element = context.element();
      // TODO: Logging every exception might overload Google Cloud Logging API. Find a better way to
      // log these errors.
      LOG.error(element.toString() + element.getErrorMessage() + element.getStacktrace());
      context.output(element);
    }
  }
}

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

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