מאפייני הביצועים של צינורות עיבוד נתונים מ-Kafka ל-BigQuery

בדף הזה מתוארות מאפייני הביצועים של משימות סטרימינג ב-Dataflow שקוראות מ-Apache Kafka וכותבות ל-BigQuery. היא מספקת תוצאות של בדיקות השוואה לצינורות map-only, שמבצעים טרנספורמציות לכל הודעה בלי לעקוב אחרי מצב או לקבץ רכיבים בזרם.

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

מתודולוגיית הבדיקה

ההשוואות בוצעו באמצעות המשאבים הבאים:

  • קלאסטר של שירות מנוהל ל-Apache Kafka. ההודעות נוצרו באמצעות התבנית Streaming Data Generator.

    • קצב העברת הודעות: בערך מיליון הודעות בשנייה
    • עומס קלט: 1 GiB/s
    • פורמט ההודעה: טקסט JSON שנוצר באופן אקראי עם סכימה קבועה
    • גודל ההודעה: בערך ‎1 KiB לכל הודעה
    • מחיצות Kafka: 1,000

  • טבלת BigQuery רגילה.

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

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

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

המדד הזה משתמש בצינור (pipeline) של מיפוי בלבד, שמבצע מיפוי פשוט והמרה של הודעות JSON. הצינור נבדק באמצעות מצב בדיוק פעם אחת ומצב של פעם אחת לפחות. עיבוד של לפחות פעם אחת מספק תפוקה טובה יותר. עם זאת, צריך להשתמש בו רק אם אפשר לקבל רשומות כפולות או אם יעד הנתונים מטפל בהסרת כפילויות.

הגדרת משרה

בטבלה הבאה אפשר לראות איך הוגדרו משימות Dataflow.

הגדרה ערך
סוג המכונה של העובד e2-standard-2
מעבדים וירטואליים (vCPU) במכונת העובד 2
זיכרון RAM של מכונת העובד 8 GB
דיסק אחסון מתמיד (persistent disk) של מכונת Worker דיסק מתמיד סטנדרטי (HDD), ‏ 30GB
מספר העובדים המקסימלי 120
מנוע סטרימינג כן
התאמה אופקית לעומס (horizontal autoscaling) כן
מודל חיוב חיוב לפי משאבים
האם Storage Write API מופעל? כן
זרמי נתונים של Storage Write API 400
תדירות ההפעלה של Storage Write API ‫5 שניות
פורמט ההודעה JSON
מצב אימות של Kafka

‫ Application Default Credentials ‏ (ADC).

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

מומלץ להשתמש ב-BigQuery Storage Write API בצינורות להזרמת נתונים. כשמשתמשים במצב 'פעם אחת בלבד' עם Storage Write API, אפשר לשנות את ההגדרות הבאות:

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

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

תוצאות ההשוואה לשוק

בקטע הזה מתוארות התוצאות של בדיקות ההשוואה לשוק.

תפוקה ושימוש במשאבים

בטבלה הבאה מוצגות תוצאות הבדיקה של קצב העברת הנתונים בצינור ושל השימוש במשאבים.

תוצאה בדיוק פעם אחת At-least-once
קצב העברת נתונים של קלט לכל עובד ממוצע: ‎15 MBps, n=3 ממוצע: ‎18 MBps, n=3
ממוצע השימוש במעבד בכל העובדים ממוצע: 70%, n=3 ממוצע: 75%, ‏ n=3
מספר צמתי העובדים ממוצע: 63, n=3 ממוצע: 53, n=3
יחידות מחשוב של Streaming Engine לשעה ממוצע: 58, n=3 ממוצע: 0, n=3

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

זמן אחזור

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

זמן האחזור הכולל מקצה לקצה של השלב, לא כולל שלב הקלט פעם אחת בלבד
P50 ממוצע: 1,200 אלפיות השנייה, n=3
P95 ממוצע: 3,000 אלפיות השנייה, n=3
P99 ממוצע: 5,400 אלפיות השנייה, n=3

הבדיקות מדדו את זמן האחזור מקצה לקצה בכל שלב (המדד job/streaming_engine/stage_end_to_end_latencies) בשלושה תהליכי בדיקה ארוכים. המדד הזה מודד כמה זמן מנוע הסטרימינג מבלה בכל שלב בצינור. הוא כולל את כל השלבים הפנימיים של הצינור, כמו:

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

בגלל מגבלה של המדד, זמן האחזור של שלב הקלט לא מדווח. לכן, הוא לא נכלל בסכום הכולל.

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

הערכת עלויות

כדי להעריך את עלות הבסיס של צינור השוואה משלכם עם חיוב לפי משאבים, אתם יכולים להשתמש בGoogle Cloud מחשבון התמחור באופן הבא:

  1. פותחים את מחשבון העלויות.
  2. לוחצים על הוספה לאומדן.
  3. בוחרים באפשרות Dataflow.
  4. בקטע סוג השירות, בוחרים באפשרות Dataflow Classic.
  5. לוחצים על הגדרות מתקדמות כדי לראות את כל האפשרויות.
  6. בוחרים את המיקום שבו העבודה תפעל.
  7. בקטע סוג העבודה, בוחרים באפשרות 'סטרימינג'.
  8. בוחרים באפשרות הפעלת מנוע הסטרימינג.
  9. מזינים את המידע על שעות הפעלת העבודה, צמתי העובדים, מכונות העובדים ואחסון Persistent Disk.
  10. מזינים את המספר המשוער של יחידות החישוב של Streaming Engine.

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

לדוגמה, נניח שאתם מפעילים צינור רק של מיפוי במצב 'פעם אחת בדיוק', עם קצב נתוני קלט של 100 MiB/s. על סמך תוצאות הבדיקה של צינור עיבוד נתונים של ‎1 GiB/s, אפשר לאמוד את דרישות המשאבים באופן הבא:

  • גורם לקביעת קנה מידה: ‎(100 MiB/s) / (1 GiB/s) = 0.1
  • צמתי עובדים משוערים: 63 עובדים × 0.1 = 6.3 עובדים
  • מספר משוער של יחידות חישוב של Streaming Engine לשעה: 58 × 0.1 = 5.8 יחידות לשעה

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

הרצת צינור לבדיקה

בקטע הזה מוצגות הפקודות gcloud dataflow flex-template run ששימשו להפעלת צינור הנתונים של המפה בלבד.

מצב 'פעם אחת בדיוק' 

gcloud dataflow flex-template run JOB_NAME \
  --project=PROJECT_ID \
  --template-file-gcs-location=gs://dataflow-templates-us-central1/latest/flex/Kafka_to_BigQuery_Flex \
  --enable-streaming-engine \
  --parameters \
readBootstrapServerAndTopic="KAFKA_BOOTSTRAP_ADDRESS;KAFKA_TOPIC",\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
messageFormat=JSON,\
writeMode=SINGLE_TABLE_NAME,\
outputTableSpec="PROJECT_ID:BQ_DATASET.BQ_TABLE_NAME",\
useBigQueryDLQ=true,\
outputDeadletterTable="PROJECT_ID:BQ_DATASET.BQ_TABLE_NAME_dlq",\
numStorageWriteApiStreams=400

מצב 'לפחות פעם אחת' 

gcloud dataflow flex-template run JOB_NAME \
  --project=PROJECT_ID \
  --template-file-gcs-location=gs://dataflow-templates-us-central1/latest/flex/Kafka_to_BigQuery_Flex \
  --enable-streaming-engine \
  --additional-experiments=streaming_mode_at_least_once \
  --parameters \
readBootstrapServerAndTopic="KAFKA_BOOTSTRAP_ADDRESS;KAFKA_TOPIC",\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
messageFormat=JSON,\
writeMode=SINGLE_TABLE_NAME,\
outputTableSpec="PROJECT_ID:BQ_DATASET.BQ_TABLE_NAME",\
useBigQueryDLQ=true,\
outputDeadletterTable="PROJECT_ID:BQ_DATASET.BQ_TABLE_NAME_dlq",\
numStorageWriteApiStreams=400,\
useStorageWriteApiAtLeastOnce=true

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

  • ‫JOB_NAME: שם המשימה ב-Dataflow
  • ‫PROJECT_ID: מזהה הפרויקט
  • ‫KAFKA_BOOTSTRAP_ADDRESS: כתובת ה-bootstrap של אשכול Apache Kafka
  • ‫KAFKA_TOPIC: השם של נושא Kafka
  • ‫BQ_DATASET: השם של מערך הנתונים ב-BigQuery
  • ‫BQ_TABLE_NAME: השם של הטבלה ב-BigQuery

יצירת נתונים לבדיקה

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

gcloud dataflow flex-template run JOB_NAME \
  --project=PROJECT_ID \
  --template-file-gcs-location=gs://dataflow-templates-us-central1/latest/flex/Streaming_Data_Generator \
  --max-workers=140 \
  --parameters \
schemaLocation=SCHEMA_LOCATION,\
qps=1000000,\
sinkType=KAFKA,\
bootstrapServer=KAFKA_BOOTSTRAP_ADDRESS,\
kafkaTopic=KAFKA_TOPIC,\
outputType=JSON

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

  • ‫JOB_NAME: שם המשימה ב-Dataflow
  • ‫PROJECT_ID: מזהה הפרויקט
  • ‫SCHEMA_LOCATION: הנתיב לקובץ סכימה ב-Cloud Storage
  • ‫KAFKA_BOOTSTRAP_ADDRESS: כתובת ה-bootstrap של אשכול Apache Kafka
  • ‫KAFKA_TOPIC: השם של נושא Kafka

תבנית מחולל נתוני הסטרימינג משתמשת בקובץ מחולל נתוני JSON כדי להגדיר את סכימת ההודעות. במבחני ההשוואה נעשה שימוש בסכימת הודעות שדומה לזו שמוצגת בהמשך:

{
  "logStreamId": "{{integer(1000001,2000000)}}",
  "message": "{{alphaNumeric(962)}}"
}

השלבים הבאים