קלט/פלט מנוהל של Dataflow ל-Apache Iceberg

Managed I/O תומך ביכולות הבאות של Apache Iceberg:

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

כדי להשתמש בטבלאות BigQuery ל-Apache Iceberg, צריך להשתמש במחבר BigQueryIO עם BigQuery Storage API. הטבלה חייבת להיות קיימת; יצירת טבלה דינמית לא אפשרית.

דרישות

ערכות ה-SDK הבאות תומכות ב-I/O מנוהל ל-Apache Iceberg:

  • ‫Apache Beam SDK for Java בגרסה 2.58.0 ואילך
  • ‫Apache Beam SDK ל-Python בגרסה 2.61.0 ואילך

הגדרות אישיות

‫Managed I/O for Apache Iceberg תומך בפרמטרים הבאים של ההגדרה:

ICEBERG קריאה

הגדרות אישיות סוג תיאור
table str מזהה של טבלת Iceberg.
catalog_name str השם של הקטלוג שמכיל את הטבלה.
catalog_properties map[str, str] מאפיינים שמשמשים להגדרת קטלוג Iceberg.
config_properties map[str, str] מאפיינים שמועברים להגדרות של Hadoop.
ירידה list[str] קבוצת משנה של שמות עמודות שרוצים להחריג מקריאה. אם הערך הוא null או ריק, כל העמודות ייקראו.
מסנ str פרדיקט שדומה ל-SQL לסינון נתונים בזמן הסריקה. דוגמה: "id > 5 AND status = 'ACTIVE'". משתמש בתחביר של Apache Calcite: ‏ https://calcite.apache.org/docs/reference.html
keep list[str] קבוצת משנה של שמות עמודות לקריאה בלבד. אם הערך הוא null או ריק, כל העמודות ייקראו.

ICEBERG כתיבה

').

הגדרות אישיות סוג תיאור
table str מזהה טבלה מוגדר במלואו. אפשר גם לספק תבנית לכתיבה למספר יעדים דינמיים, לדוגמה: ‎`dataset.my_{col1}_{col2.nested}_table`‎.
allowed_lateness_seconds int32 כמה זמן יכול לעבור בין הזמן שבו רשומה מאוחרת מגיעה לבין הזמן שבו היא נמחקת לגמרי, במקום להיות מנותבת לפלט dead_letter. ברירת המחדל היא 21,600 (6 שעות). בשלב הזה יש תמיכה רק במצב 'מיזוג בקריאה'.
חלוקת נתונים אוטומטית boolean ההגדרה הזו מאפשרת חלוקה דינמית למקטעים, כדי לשנות באופן אוטומטי את מספר הכותבים המקבילים על סמך נפח הנתונים. הוא מטפל בחלוקת נתונים לא מאוזנת על ידי חלוקה נוספת של המחיצות לכמה מחיצות Shard, כדי למנוע צווארי בקבוק במהלך כתיבה של נתונים בהיקף גדול. האפשרות זמינה רק במצב ההפצה 'hash'.
catalog_name str השם של הקטלוג שמכיל את הטבלה.
catalog_properties map[str, str] מאפיינים שמשמשים להגדרת קטלוג Iceberg.
change_type_column str מיזוג בזמן קריאה בלבד. שם העמודה האופציונלית שמייצגת את סוג השינוי בשורה (INSERT,‏ UPDATE_BEFORE,‏ UPDATE_AFTER או DELETE). העמודה הזו תוסר משורת הנתונים לפני הכתיבה ל-Iceberg. אם לא מוגדר, יעד הנתונים ישתמש ב-ValueKind המקורי של הרכיב
change_type_map map[str, str] מיזוג בזמן קריאה בלבד. אופציונלי: מיפוי מערך של change_type_column לשם הקנוני של סוג השינוי (ראו למעלה).
config_properties map[str, str] מאפיינים שמועברים להגדרות של Hadoop.
direct_write_byte_limit int32 בצינור עיבוד נתונים (pipeline) של סטרימינג, מגדיר את המגבלה להעברת חבילות לנתיב הכתיבה הישיר.
distribution_mode str הגדרה של הפצת נתוני כתיבה. הפצות נתמכות: - none: לא משנים את הסדר של השורות (ברירת מחדל) - hash: משנים את הסדר של השורות לפי מפתח המחיצה לפני כתיבת הנתונים
ירידה list[str] רשימה של שמות שדות שיוסרו מרשומת הקלט לפני הכתיבה. הערך הזה מוציא מכלל אפשרות את הערכים keep ו-only. במצב מיזוג בקריאה, עמודות הבקרה תמיד מושמטות.
equality_columns list[str] עמודות שמגדירות את זהות השורה (שדות של שוויון-מחיקה). ברירת המחדל היא שדות המזהה (המפתח הראשי) של טבלת היעד. חובה אם הטבלה עדיין לא קיימת. בשלב הזה יש תמיכה רק במצב 'מיזוג בקריאה'.
keep list[str] רשימה של שמות שדות שיישארו ברשומת הקלט. כל שאר השדות יימחקו לפני הכתיבה. הערך הזה לא יכול להיות ביחד עם הערכים drop ו-only. במצב מיזוג בזמן קריאה, עמודות הבקרה מושמטות אלא אם הן מופיעות כאן.
maximum_table_cache_size int32 בצינורות להעברת נתונים באצווה, מגדיר את המספר המקסימלי של מפרטי מטא-נתונים של טבלאות שיישארו במטמון בזיכרון. אם הטבלה חורגת מהמגבלה הזו, המערכת חוזרת לטעינת קטלוג מקומי של העובד.
מצב str קובעת איך השורות נכתבות. ‫'append' (הוספה) (ברירת מחדל) מוסיף כל שורה כנתונים חדשים. האפשרות 'מיזוג בקריאה' מתייחסת לכל שורה כאל שינוי (INSERT,‏ UPDATE_BEFORE,‏ UPDATE_AFTER או DELETE) שמוחל על הטבלה לפי המפתח הראשי.
num_shards int32 מספר הרסיסים של הגיבוב של המפתח הראשי הדטרמיניסטי לכל יעד, כלומר מקסימום מקביליות הכתיבה לכל יעד. ערך נמוך מדי עלול לגרום לצוואר בקבוק בכתיבה, וערך גבוה מדי עלול ליצור יותר קבצים. ברירת המחדל היא 16. בשלב הזה יש תמיכה רק במצב 'מיזוג בקריאה'.
רק str השם של שדה רשומה יחיד שצריך לכתוב. האפשרות הזו לא יכולה להיות משולבת עם האפשרויות 'שמירה' ו 'הסרה'.
partition_fields list[str] שדות שמשמשים ליצירת מפרט מחיצה שמוחל כשיוצרים טבלאות. עבור השדה foo, הטרנספורמציות הזמינות של המחיצה הן:
  • foo
  • truncate(foo, N)
  • bucket(foo, N)
  • hour(foo)
  • day(foo)
  • month(foo)
  • year(foo)
  • void(foo)

מידע נוסף על טרנספורמציות של מחיצות זמין בכתובת https://iceberg.apache.org/spec/#partition-transforms.

sequence_number_column str מיזוג בזמן קריאה בלבד. שם העמודה הנדרש שמייצג את המספר הסידורי המונוטוני שמשמש לסידור השינויים של מפתח יחיד. ברירת המחדל היא ‎ '_commit_snapshot_sequence_number'‎. העמודה הזו תוסר משורת הנתונים לפני הכתיבה ל-Iceberg.
shards_per_partition int32 המספר המקסימלי של רסיסים ששורות של מחיצה אחת יכולות לתפוס. ערכים נמוכים יותר כותבים פחות קבצים לכל פעולת commit, אבל גם מצמצמים את המקביליות של הכתיבה לכל מחיצה. ערך של 1 מצמיד כל מחיצה לכותב אחד. ההגדרה הזו לא רלוונטית לטבלאות לא מחולקות. הערך צריך להיות בין 1 ל-num_shards. ערך ברירת המחדל הוא num_shards. בשלב הזה יש תמיכה רק במצב 'מיזוג בקריאה'.
sink_id str מזהה קבוע של יעד הנתונים הזה, שמשמש למרחב שמות של טוקנים של אידמפוטנטיות שנכתבים לסיכום של תמונת מצב של כל קומיט ב-Iceberg. ברירת המחדל היא מזהה UUID ייחודי לכל פעולת כתיבה. להגדיר אותו באופן מפורש (ולשמור על יציבות שלו בהפעלות מחדש) לביצועים של בדיוק פעם אחת בהפעלות מחדש של כתיבה מסוימת לסטרימינג. טעינת נתונים באצווה עם sink_id יציב מתבצעת רק פעם אחת (טעינות נתונים באצווה מאוחרות יותר עם אותו sink_id מדלגות). בשלב הזה יש תמיכה רק במצב 'מיזוג בקריאה'.
snapshot_properties map[str, str] מאפייני מפתח/ערך נוספים שיוגדרו בכל סיכום של תמונת מצב של Iceberg. מפתחות עם הקידומת beam.cdc. שמורים ויידחו. בשלב הזה יש תמיכה רק במצב 'מיזוג בקריאה'.
sort_fields list[str] השדות שמשמשים להגדרת סדר המיון של הטבלה, שמוחל כשהטבלה נוצרת. כל רשומה היא מהצורה <term> [asc|desc] [nulls first|nulls last], כאשר <term> הוא שם שדה או אחת מהטרנספורמציות של החלוקה למחיצות (למשל bucket(col, 4),‏ day(ts)). כברירת מחדל, הכיוון הוא עולה, והסדר של ערכי null הוא nulls-first בסדר עולה ו-nulls-last בסדר יורד. הערה: הפעולה הזו מגדירה את סדר המיון המוצהר של הטבלה כמטא-נתונים, אבל היא לא גורמת ל-Beam למיין פיזית את הרשומות לפני הכתיבה. מידע נוסף על סדר מיון זמין בכתובת https://iceberg.apache.org/spec/#sort-orders.
sorter_memory_mb int32 גודל שטח האחסון הזמני בזיכרון (MB) למיון לפני הכתיבה. קבוצות גדולות יותר מזה נשפכות לדיסק. הערך חייב להיות ‎ >= 1. ברירת המחדל היא 100. בשלב הזה יש תמיכה רק במצב 'מיזוג בקריאה'.
table_cache_polling_buckets int32 מגדיר את מספר הדליים או העובדים המקבילים שמשמשים לשליחת שאילתות לקטלוג Iceberg במהלך רענון. ברירת המחדל היא 1.
table_cache_refresh_interval_seconds int32 בצינורות להעברת נתונים בסטרימינג, המערכת מגדירה את המרווח בשניות שבו מתבצע רענון של מטא-נתונים של טבלה מהקטלוג.
table_properties map[str, str] מאפייני טבלה של Iceberg שיוגדרו בטבלה כשהיא תיווצר. מידע נוסף על מאפייני טבלה זמין בכתובת https://iceberg.apache.org/docs/latest/configuration/#table-properties.
token_heartbeat_seconds int32 סטרימינג בלבד. אם ההגדרה הזו מופעלת, ה-sink ישלח מדי פעם התחייבות ריקה לרענון האסימון בזמן שהוא לא פעיל, כך שהשרשור של תמונת המצב עם חותמת הזמן sink_id יישאר עדכני ופחות סביר שהוא יאבד בגלל expire_snapshots. ההגדרה מושבתת כברירת מחדל. בשלב הזה יש תמיכה רק במצב 'מיזוג בקריאה'.
triggering_frequency_seconds int32 בצינור עיבוד נתונים (pipeline) של סטרימינג, ההגדרה הזו קובעת את התדירות שבה נוצרות תמונות מצב.
upsert boolean מיזוג בזמן קריאה בלבד. אם הערך הוא true, רק התמונה שאחרי כל שינוי (INSERT/UPDATE_AFTER) מוחלת כ-upsert, ורשומות UPDATE_BEFORE מושמטות. ברירת מחדל: false.
use_side_input_table_cache boolean מאפשר אחסון במטמון של מטא-נתונים של טבלאות Iceberg כקלט צדדי שפג תוקפו אצל כל העובדים, כדי להפחית את עומס הקטלוג.
write_properties map[str, str] מאפיינים שמוחלים על כלי הכתיבה של הקובץ הבסיסי (למשל, מאפייני כתיבה של Parquet כמו 'write.parquet.bloom-filter-enabled.column.

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

מידע נוסף ודוגמאות קוד זמינים בנושאים הבאים: