קריאה מ-Apache Kafka אל Dataflow

במסמך הזה מוסבר איך לקרוא נתונים מ-Apache Kafka ל-Dataflow, והוא כולל טיפים לשיפור הביצועים ושיטות מומלצות.

ברוב תרחישי השימוש, מומלץ להשתמש במחבר I/O מנוהל כדי לקרוא מ-Kafka.

אם אתם צריכים כוונון ביצועים מתקדם יותר, כדאי להשתמש במחבר KafkaIO. מחבר KafkaIO זמין ל-Java או באמצעות מסגרת צינורות רב-לשוניים ל-Python ול-Go.

מקביליות

בקטעים הבאים מוסבר איך להגדיר מקביליות כשקוראים מ-Kafka.

סקירה כללית

המקביליות מוגבלת על ידי שני גורמים: מספר העובדים המקסימלי (max_num_workers) ומספר המחיצות של Kafka. ב-Dataflow, ערך ברירת המחדל של פיצול מקבילי הוא 4 כפול max_num_workers. עם זאת, הפיצול מוגבל במספר המחיצות. לדוגמה, אם יש 100 מעבדים וירטואליים זמינים, אבל צינור הנתונים קורא רק מ-10 מחיצות של Kafka, רמת המקביליות המקסימלית היא 10.

כדי למקסם את המקביליות, מומלץ להשתמש ב-4 מחיצות Kafka לפחות.max_num_workers אם העבודה שלכם משתמשת ב-Portable Runner, כדאי להגדיר ערך גבוה עוד יותר של מקביליות. נקודת התחלה טובה היא ליצור מחיצות ששווה למספר ליבות ה-vCPU של העובד כפול 2.

הפצה מחדש

אם אי אפשר להגדיל את מספר המחיצות, אפשר להגדיל את המקביליות על ידי קריאה ל-KafkaIO.Read.withRedistribute. השיטה הזו מוסיפה טרנספורמציה Redistribute לצינור, שמספקת רמז ל-Dataflow כדי לבצע מחדש חלוקה והקבלה של הנתונים בצורה יעילה יותר. מומלץ מאוד לציין את מספר הרסיסים האופטימלי על ידי קריאה ל-KafkaIO.Read.withRedistributeNumKeys. שימוש ב-KafkaIO.Read.withRedistribute בלבד יכול ליצור מספר רב של מפתחות, מה שיוביל לבעיות בביצועים. מידע נוסף זמין במאמר בנושא זיהוי שלבים עם מקביליות גבוהה. הפצה מחדש של הנתונים מוסיפה תקורה מסוימת לביצוע שלב הערבוב. מידע נוסף זמין במאמר בנושא מניעת מיזוג.

כדי לצמצם את העלות של ערבוב מחדש של חלוקה, צריך להתקשר אל KafkaIO.Read.withOffsetDeduplication. במצב הזה, כמות הנתונים שצריך לשמור כחלק מהערבוב מצטמצמת, אבל עדיין מתבצע עיבוד מדויק של כל נתון.

אם לא נדרש עיבוד של כל הודעה בדיוק פעם אחת, אפשר לאפשר כפילויות על ידי קריאה ל-KafkaIO.Read.withAllowDuplicates.

בטבלה הבאה מפורטות האפשרויות להפצה מחדש:

אפשרות מצב עיבוד Apache Beam הגדרות אישיות
חלוקה מחדש של הקלט בדיוק פעם אחת גרסה 2.60 ואילך KafkaIO.Read.withRedistribute()
אפשרות להוסיף כפילויות At-least-once גרסה 2.60 ואילך KafkaIO.Read.withRedistribute().withAllowDuplicates()
ביטול כפילויות בהיסט בדיוק פעם אחת גרסה 2.69 ואילך KafkaIO.Read.withRedistribute().withOffsetDeduplication()

הטיה בעומס

כדאי לוודא שהעומס בין המחיצות מאוזן יחסית ולא מוטה. אם העומס לא מאוזן, יכול להיות שהעובדים לא ינוצלו בצורה טובה. עובדים שקוראים ממחיצות עם עומס קל יותר עשויים להיות יחסית לא פעילים, ועובדים שקוראים ממחיצות עם עומס כבד עשויים להישאר מאחור. ‫Dataflow מספק מדדים לגבי ה-backlog לכל מחיצה.

אם העומס לא מאוזן, איזון דינמי של העבודה יכול לעזור לחלק את העבודה. לדוגמה, יכול להיות ש-Dataflow יקצה worker אחד לקריאה ממספר מחיצות עם נפח נמוך, ויקצה worker אחר לקריאה ממחיצה אחת עם נפח גבוה. עם זאת, שני עובדים לא יכולים לקרוא מאותה מחיצה, ולכן מחיצה שעמוסה מאוד עדיין יכולה לגרום לעיכוב בצינור.

שיטות מומלצות

בקטע הזה מפורטות המלצות לקריאה מ-Kafka אל Dataflow.

נושאים עם נפח חיפוש נמוך

תרחיש נפוץ הוא קריאה ממספר רב של נושאים עם נפח נמוך בו-זמנית – לדוגמה, נושא אחד לכל לקוח. יצירת משימות נפרדות של Dataflow לכל נושא היא לא חסכונית, כי כל משימה דורשת לפחות עובד מלא אחד. במקום זאת, אפשר לנסות את האפשרויות הבאות:

  • מיזוג נושאים. לשלב נושאים לפני שהם מוזנים ל-Dataflow. עדיף להוסיף כמה נושאים עם נפח חיפוש גבוה מאשר הרבה נושאים עם נפח חיפוש נמוך. כל נושא עם נפח גבוה יכול להיות מטופל על ידי משימת Dataflow אחת שמנצלת באופן מלא את העובדים שלה.

  • קריאת כמה נושאים אם אי אפשר לשלב נושאים לפני ההטמעה שלהם ב-Dataflow, כדאי ליצור צינור שקורא מנושאים מרובים. הגישה הזו מאפשרת ל-Dataflow להקצות כמה נושאים לאותה מכונת worker. יש שתי דרכים להטמיע את הגישה הזו:

    • שלב קריאה יחיד. יוצרים מופע יחיד של מחבר KafkaIO ומגדירים אותו לקריאת מספר נושאים. אחר כך מסננים לפי שם הנושא כדי להחיל לוגיקה שונה לכל נושא. דוגמאות לקוד זמינות במאמר בנושא קריאה מכמה נושאים. כדאי להשתמש באפשרות הזו אם כל הנושאים נמצאים באותו אשכול. חיסרון אחד הוא שבעיות ב-sink או ב-transform יחידים עלולות לגרום להצטברות של נתונים שלא נשלחו בכל הנושאים.

      בתרחישי שימוש מתקדמים יותר, אפשר להעביר קבוצה של KafkaSourceDescriptorאובייקטים שמציינים את הנושאים לקריאה. אם משתמשים באפשרות KafkaSourceDescriptor, אפשר לעדכן את רשימת הנושאים בהמשך, אם יש צורך. כדי להשתמש בתכונה הזו, צריך Java עם Portable Runner.

    • Multiple read steps (קריאה של נתוני הצעדים). כדי לקרוא מנושאים שנמצאים באשכולות שונים, אפשר לכלול בצינור כמה מופעים של KafkaIO. בזמן שהעבודה פועלת, אפשר לעדכן מקורות בודדים באמצעות מיפויי טרנספורמציה. הגדרת נושא או מקבץ חדשים נתמכת רק כשמשתמשים ב-Portable Runner. הגישה הזו עלולה להקשות על מעקב, כי צריך לעקוב אחרי כל טרנספורמציה של קריאה בנפרד במקום להסתמך על מדדים ברמת צינור הנתונים.

ביצוע Commit חזרה ל-Kafka

כברירת מחדל, מחבר KafkaIO לא משתמש בהיסטים של Kafka כדי לעקוב אחרי ההתקדמות ולא מבצע commit בחזרה ל-Kafka. אם מתקשרים אל commitOffsetsInFinalize, המחבר עושה כמיטב יכולתו לבצע commit בחזרה ל-Kafka אחרי שרשומות מבוצעות ב-Dataflow. יכול להיות שרשומות שבוצעו ב-Dataflow לא יעברו עיבוד מלא, ולכן אם מבטלים את צינור הנתונים, יכול להיות שקיזוז יבוצע בלי שהרשומות יעברו עיבוד מלא.

לא מומלץ להשתמש באפשרות הזו כי הגדרת enable.auto.commit=True מבצעת commit להיסטים ברגע שהם נקראים מ-Kafka בלי ש-Dataflow יעבד אותם. ההמלצה היא להגדיר גם את enable.auto.commit=False וגם את commitOffsetsInFinalize=True. אם מגדירים את enable.auto.commit ל-True, יכול להיות שהנתונים יאבדו אם הצינור ייקטע במהלך העיבוד. יכול להיות שרשומות שכבר בוצעו ב-Kafka יושמטו.

סימני מים

כברירת מחדל, המחבר KafkaIO משתמש בזמן העיבוד הנוכחי כדי להקצות את סימן המים של הפלט ואת זמן האירוע. כדי לשנות את ההתנהגות הזו, קוראים ל-withTimestampPolicyFactory ומקצים TimestampPolicy. ‫Beam מספקת הטמעות של TimestampPolicy שמחשבות את סימן המים על סמך זמן הוספת היומן של Kafka או שעת יצירת ההודעה.

דברים שחשוב לזכור לגבי ראנרים

למחבר KafkaIO יש שתי הטמעות בסיסיות לקריאות של Kafka: הגרסה הישנה יותר ReadFromKafkaViaUnbounded והגרסה החדשה יותר ReadFromKafkaViaSDF. ‫Dataflow בוחר באופן אוטומטי את ההטמעה הכי טובה למשימה על סמך שפת ה-SDK ודרישות המשימה. אל תבקשו במפורש להשתמש ב-Runner או ביישום של Kafka, אלא אם אתם צריכים תכונות ספציפיות שזמינות רק ביישום הזה. מידע נוסף על בחירת רץ זמין במאמר בנושא שימוש ב-Dataflow Portable Runner.

אם הצינור שלכם משתמש ב-withTopic או ב-withTopics, ההטמעה הישנה יותר שולחת שאילתה ל-Kafka בזמן בניית הצינור כדי לקבל את המחיצות הזמינות. למכונה שיוצרת את צינור הנתונים צריכה להיות הרשאה להתחבר ל-Kafka. אם מופיעה שגיאת הרשאה, צריך לוודא שיש לכם הרשאות להתחבר ל-Kafka באופן מקומי. כדי להימנע מהבעיה הזו, אפשר להשתמש ב-withTopicPartitions, שלא מתחבר ל-Kafka בזמן בניית צינור הנתונים.

פריסה בסביבת הייצור

כשפורסים את הפתרון בסביבת ייצור, מומלץ להשתמש בתבניות גמישות. שימוש בתבנית Flex מאפשר להפעיל את צינור הנתונים מסביבה עקבית, מה שיכול לעזור לצמצם בעיות בהגדרות המקומיות.

הרישום ביומן מ-KafkaIO יכול להיות מפורט מאוד. כדאי לשקול להפחית את רמת הרישום ביומן בייצור באופן הבא:

sdkHarnessLogLevelOverrides='{"org.apache.kafka.clients.consumer.internals.SubscriptionState":"WARN"}'.

מידע נוסף זמין במאמר בנושא הגדרת רמות יומן של עובדי צינורות.

הגדרה של רשתות

כברירת מחדל, ‏ Dataflow מפעיל מופעים ברשת ברירת המחדל של הענן הווירטואלי הפרטי (VPC). בהתאם להגדרות של Kafka, יכול להיות שתצטרכו להגדיר רשת ורשת משנה שונות עבור Dataflow. מידע נוסף זמין במאמר ציון רשת ורשת משנה. כשמגדירים את הרשת, צריך ליצור כללי חומת אש שמאפשרים למכונות העובד של Dataflow להגיע לשרתי Kafka.

אם אתם משתמשים ב-VPC Service Controls, אתם צריכים למקם את אשכול Kafka בתוך גבולות הגזרה של VPC Service Controls, או להרחיב את גבולות הגזרה ל-VPN או ל-Cloud Interconnect המורשים.

אם אשכול Kafka נפרס מחוץ ל- Google Cloud, צריך ליצור חיבור רשת בין Dataflow לבין אשכול Kafka. יש כמה אפשרויות שונות לניהול רשתות, שלכל אחת מהן יש יתרונות וחסרונות:

‫Dedicated Interconnect היא האפשרות הכי טובה לביצועים ולמהימנות צפויים, אבל יכול לקחת יותר זמן להגדיר אותה כי צדדים שלישיים צריכים להקצות את המעגלים החדשים. בטופולוגיה שמבוססת על כתובת IP ציבורית, אפשר להתחיל במהירות כי לא צריך לבצע הרבה עבודות ברשת.

בקטעים הבאים מפורטות האפשרויות האלה.

מרחב כתובות RFC 1918 משותף

גם Dedicated Interconnect וגם IPsec VPN מאפשרים לכם גישה ישירה לכתובות IP של RFC 1918 בענן הווירטואלי הפרטי (VPC), מה שיכול לפשט את ההגדרה של Kafka. אם אתם משתמשים בטופולוגיה שמבוססת על VPN, כדאי להגדיר VPN עם תפוקה גבוהה.

כברירת מחדל, Dataflow מפעיל אינסטנסים ברשת ה-VPC שמוגדרת כברירת מחדל. בטופולוגיה של רשת פרטית עם מסלולים שמוגדרים באופן מפורש ב-Cloud Router שמקשרים בין רשתות משנה ב- Google Cloud לאותו אשכול Kafka, אתם צריכים יותר שליטה במיקום של מופעי Dataflow. אפשר להשתמש ב-Dataflow כדי להגדיר את network וsubnetwork פרמטרים של ביצוע.

חשוב לוודא שלרשת המשנה המתאימה יש מספיק כתובות IP זמינות כדי ש-Dataflow יוכל להפעיל מופעים כשהוא מנסה להרחיב את קנה המידה. בנוסף, כשיוצרים רשת נפרדת להפעלת מופעי Dataflow, צריך לוודא שיש כלל של חומת אש שמאפשר תעבורת TCP בין כל המכונות הווירטואליות בפרויקט. כלל חומת האש הזה כבר מוגדר ברשת שמוגדרת כברירת מחדל.

מרחב כתובות IP ציבוריות

הארכיטקטורה הזו משתמשת בפרוטוקול Transport Layer Security ‏(TLS) כדי לאבטח את התעבורה בין לקוחות חיצוניים לבין Kafka, ומשתמשת בתעבורה לא מוצפנת לתקשורת בין ברוקרים. אם מאזין Kafka נקשר לממשק רשת שמשמש לתקשורת פנימית וחיצונית, קל להגדיר את המאזין. עם זאת, בתרחישים רבים, הכתובות שמפורסמות חיצונית של ברוקרי Kafka באשכול שונות מממשקי הרשת הפנימיים שבהם Kafka משתמש. בתרחישים כאלה, אפשר להשתמש במאפיין advertised.listeners:

# Configure protocol map
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:SSL

# Use plaintext for inter-broker communication
inter.broker.listener.name=INTERNAL

# Specify that Kafka listeners should bind to all local interfaces
listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:9093

# Separately, specify externally visible address
advertised.listeners=INTERNAL://kafkabroker-n.mydomain.com:9092,EXTERNAL://kafkabroker-n.mydomain.com:9093

לקוחות חיצוניים מתחברים באמצעות יציאה 9093 דרך ערוץ SSL, ולקוחות פנימיים מתחברים באמצעות יציאה 9092 דרך ערוץ טקסט רגיל. כשמציינים כתובת בקטע advertised.listeners, צריך להשתמש בשמות DNS (kafkabroker-n.mydomain.com בדוגמה הזו) שמפנים לאותו מופע גם לתנועה חיצונית וגם לתנועה פנימית. יכול להיות ששימוש בכתובות IP ציבוריות לא יפעל כי יכול להיות שהכתובות לא יצליחו לפתור בעיות בתנועה פנימית.

התאמה של Kafka

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

  • unboundedReaderMaxElements. ברירת המחדל היא 10,000. ערך גבוה יותר, כמו 100,000, יכול להגדיל את הגודל של החבילות, מה שיכול לשפר משמעותית את הביצועים אם הצינור כולל צבירות. עם זאת, ערך גבוה יותר עשוי גם להגדיל את זמן האחזור. כדי להגדיר את הערך, משתמשים ב-setUnboundedReaderMaxElements. ההגדרה הזו לא חלה על Portable Runner. ב-Portable Runner, משתמשים באפשרות השירות Dataflow‏ sdf_checkpoint_after_output_bytes.

  • unboundedReaderMaxReadTimeMs. ברירת המחדל היא 10,000 אלפיות השנייה. ערך גבוה יותר, כמו 20,000 אלפיות השנייה, יכול להגדיל את גודל החבילה, בעוד שערך נמוך יותר, כמו 5,000 אלפיות השנייה, יכול להקטין את ההשהיה או את העומס. כדי להגדיר את הערך, משתמשים ב-setUnboundedReaderMaxReadTimeMs. ההגדרה הזו לא חלה על Portable Runner. ב-Portable Runner, משתמשים באפשרות השירות Dataflow‏ sdf_checkpoint_after_duration.

  • ‫max.poll.records. ברירת המחדל היא 500. יכול להיות שערך גבוה יותר יניב ביצועים טובים יותר כי הוא יאפשר אחזור של יותר רשומות נכנסות יחד, במיוחד כשמשתמשים ב-Portable Runner. כדי להגדיר את הערך, קוראים לפונקציה withConsumerConfigUpdates.

  • ‫fetch.max.bytes. ברירת המחדל היא 1MB. ערך גבוה יותר עשוי לשפר את קצב העברת הנתונים על ידי צמצום מספר הבקשות, במיוחד כשמשתמשים ב-Portable Runner. עם זאת, הגדרה של ערך גבוה מדי עלולה להגדיל את זמן האחזור, למרות שסביר יותר שהצוואר בקבוק העיקרי הוא עיבוד במורד הזרם. ערך התחלתי מומלץ הוא 100MB. כדי להגדיר את הערך, קוראים לפונקציה withConsumerConfigUpdates.

  • ‫max.partition.fetch.bytes. ברירת המחדל היא 1MB. הפרמטר הזה מגדיר את כמות הנתונים המקסימלית לכל מחיצה שהשרת מחזיר. הגדלת הערך יכולה לשפר את קצב העברת הנתונים על ידי הקטנת מספר הבקשות, במיוחד כשמשתמשים ב-Portable Runner. עם זאת, אם מגדירים ערך גבוה מדי, יכול להיות שזמן האחזור יתארך, למרות שסביר יותר שהצוואר בקבוק העיקרי הוא העיבוד במורד הזרם. הערך המומלץ להתחלה הוא 100MB. כדי להגדיר את הערך, קוראים לפונקציה withConsumerConfigUpdates.

  • ‫consumerPollingTimeout. ברירת המחדל היא 2 שניות. אם הלקוח הצרכני מגיע לזמן קצוב לתפוגה לפני שהוא יכול לקרוא רשומות, נסו להגדיר ערך גבוה יותר. ההגדרה הזו רלוונטית בעיקר כשמבצעים קריאות בין אזורים או קריאות ברשת איטית. כדי להגדיר את הערך, קוראים לפונקציה withConsumerPollingTimeout.

מוודאים ש-receive.buffer.bytes גדול מספיק כדי להכיל את גודל ההודעות. אם הערך קטן מדי, יכול להיות שבלוגים יופיעו צרכנים שנוצרים מחדש באופן רציף ומחפשים היסט ספציפי.

דוגמאות

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

קריאה מנושא יחיד

בדוגמה הזו נשתמש במחבר קלט/פלט מנוהל. המדריך כולל הסברים על קריאה מנושא ב-Kafka וכתיבה של מטעני נתונים של הודעות לקובצי טקסט.

Java

כדי לבצע אימות ב-Dataflow, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לסביבת פיתוח מקומית. 

import com.google.common.collect.ImmutableMap;
import java.io.UnsupportedEncodingException;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.managed.Managed;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;

public class KafkaRead {

  public static Pipeline createPipeline(Options options) {

    // Create configuration parameters for the Managed I/O transform.
    ImmutableMap<String, Object> config = ImmutableMap.<String, Object>builder()
        .put("bootstrap_servers", options.getBootstrapServer())
        .put("topic", options.getTopic())
        .put("format", "RAW")
        .put("max_read_time_seconds", 15)
        .put("auto_offset_reset_config", "earliest")
        .build();

    // Build the pipeline.
    var pipeline = Pipeline.create(options);
    pipeline
        // Read messages from Kafka.
        .apply(Managed.read(Managed.KAFKA).withConfig(config)).getSinglePCollection()
        // Get the payload of each message and convert to a string.
        .apply(MapElements
            .into(TypeDescriptors.strings())
            .via((row -> {
              var bytes = row.getBytes("payload");
              try {
                return new String(bytes, "UTF-8");
              } catch (UnsupportedEncodingException e) {
                throw new RuntimeException(e);
              }
            })))
        // Write the payload to a text file.
        .apply(TextIO
            .write()
            .to(options.getOutputPath())
            .withSuffix(".txt")
            .withNumShards(1));
    return pipeline;
  }
}

Python

כדי לבצע אימות ב-Dataflow, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לסביבת פיתוח מקומית. 

import argparse

import apache_beam as beam

from apache_beam import window
from apache_beam.io.textio import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions


def read_from_kafka() -> None:
    # Parse the pipeline options passed into the application. Example:
    #     --topic=$KAFKA_TOPIC --bootstrap_server=$BOOTSTRAP_SERVER
    #     --output=$CLOUD_STORAGE_BUCKET --streaming
    # For more information, see
    # https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    class MyOptions(PipelineOptions):
        @staticmethod
        def _add_argparse_args(parser: argparse.ArgumentParser) -> None:
            parser.add_argument("--topic")
            parser.add_argument("--bootstrap_server")
            parser.add_argument("--output")

    options = MyOptions()
    with beam.Pipeline(options=options) as pipeline:
        (
            pipeline
            # Read messages from an Apache Kafka topic.
            | beam.managed.Read(
                beam.managed.KAFKA,
                config={
                  "bootstrap_servers": options.bootstrap_server,
                  "topic": options.topic,
                  "data_format": "RAW",
                  "auto_offset_reset_config": "earliest",
                  # The max_read_time_seconds parameter is intended for testing.
                  # Avoid using this parameter in production.
                  "max_read_time_seconds": 5
                }
            )
            # Subdivide the output into fixed 5-second windows.
            | beam.WindowInto(window.FixedWindows(5))
            | WriteToText(
                file_path_prefix=options.output, file_name_suffix=".txt", num_shards=1
            )
        )

קריאה מכמה נושאים

בדוגמה הזו נשתמש במחבר KafkaIO. הוא מראה איך לקרוא מכמה נושאים של Kafka ולהחיל לוגיקה נפרדת של צינורות לכל נושא.

בתרחישי שימוש מתקדמים יותר, אפשר להעביר באופן דינמי קבוצה של אובייקטים מסוג KafkaSourceDescriptor, כדי לעדכן את רשימת הנושאים לקריאה. הגישה הזו דורשת Java עם Portable Runner.

Java

כדי לבצע אימות ב-Dataflow, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לסביבת פיתוח מקומית. 

import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.transforms.Filter;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.joda.time.Duration;
import org.joda.time.Instant;

public class KafkaReadTopics {

  public static Pipeline createPipeline(Options options) {
    String topic1 = options.getTopic1();
    String topic2 = options.getTopic2();

    // Build the pipeline.
    var pipeline = Pipeline.create(options);
    var allTopics = pipeline
        .apply(KafkaIO.<Long, String>read()
            .withTopics(List.of(topic1, topic2))
            .withBootstrapServers(options.getBootstrapServer())
            .withKeyDeserializer(LongDeserializer.class)
            .withValueDeserializer(StringDeserializer.class)
            .withMaxReadTime(Duration.standardSeconds(10))
            .withStartReadTime(Instant.EPOCH)
        );

    // Create separate pipeline branches for each topic.
    // The first branch filters on topic1.
    allTopics
        .apply(Filter.by(record -> record.getTopic().equals(topic1)))
        .apply(MapElements
            .into(TypeDescriptors.strings())
            .via(record -> record.getKV().getValue()))
        .apply(TextIO.write()
            .to(topic1)
            .withSuffix(".txt")
            .withNumShards(1)
        );

    // The second branch filters on topic2.
    allTopics
        .apply(Filter.by(record -> record.getTopic().equals(topic2)))
        .apply(MapElements
            .into(TypeDescriptors.strings())
            .via(record -> record.getKV().getValue()))
        .apply(TextIO.write()
            .to(topic2)
            .withSuffix(".txt")
            .withNumShards(1)
        );
    return pipeline;
  }
}

Python

כדי לבצע אימות ב-Dataflow, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לסביבת פיתוח מקומית. 

import argparse

import apache_beam as beam

from apache_beam.io.kafka import ReadFromKafka
from apache_beam.io.textio import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions


def read_from_kafka() -> None:
    # Parse the pipeline options passed into the application. Example:
    #   --bootstrap_server=$BOOTSTRAP_SERVER --output=$STORAGE_BUCKET --streaming
    # For more information, see
    # https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    class MyOptions(PipelineOptions):
        @staticmethod
        def _add_argparse_args(parser: argparse.ArgumentParser) -> None:
            parser.add_argument('--bootstrap_server')
            parser.add_argument('--output')

    options = MyOptions()
    with beam.Pipeline(options=options) as pipeline:
        # Read from two Kafka topics.
        all_topics = pipeline | ReadFromKafka(consumer_config={
                "bootstrap.servers": options.bootstrap_server
            },
            topics=["topic1", "topic2"],
            with_metadata=True,
            max_num_records=10,
            start_read_time=0
        )

        # Filter messages from one topic into one branch of the pipeline.
        (all_topics
            | beam.Filter(lambda message: message.topic == 'topic1')
            | beam.Map(lambda message: message.value.decode('utf-8'))
            | "Write topic1" >> WriteToText(
                file_path_prefix=options.output + '/topic1/output',
                file_name_suffix='.txt',
                num_shards=1))

        # Filter messages from the other topic.
        (all_topics
            | beam.Filter(lambda message: message.topic == 'topic2')
            | beam.Map(lambda message: message.value.decode('utf-8'))
            | "Write topic2" >> WriteToText(
                file_path_prefix=options.output + '/topic2/output',
                file_name_suffix='.txt',
                num_shards=1))

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