יצירת אשכול Connect

אשכול Connect מספק סביבה למחברים שעוזרים להעביר נתונים מפריסות קיימות של Kafka לאשכול של שירות מנוהל של Google Cloud ל-Apache Kafka, או להעביר נתונים מאשכול של שירות מנוהל ל-Apache Kafka לשירות אחר של Google Cloud או לאשכול אחר של Kafka. אשכול Kafka המשני יכול להיות אשכול אחר של השירות המנוהל של Google Cloud ל-Apache Kafka, אשכול בניהול עצמי או אשכול מקומי.

לפני שמתחילים

מוודאים שכבר יצרתם אשכול של שירות מנוהל ל-Apache Kafka. אתם צריכים את השם של אשכול השירות המנוהל ל-Apache Kafka שאליו יצורף אשכול Connect.

כל אשכול Connect משויך לאשכול של שירות מנוהל ל-Apache Kafka. האשכול הזה מאחסן את המצב של המחברים שפועלים באשכול Connect.

התפקידים וההרשאות שנדרשים כדי ליצור אשכול Connect

כדי לקבל את ההרשאות שנדרשות ליצירת אשכול Connect, בקשו מהאדמין להקצות לכם את התפקיד עריכת אשכולות מנוהלים של Kafka Connect (roles/managedkafka.connectClusterEditor) ב-IAM בפרויקט. כדי לקרוא הסבר על מתן תפקידים, ראו איך מנהלים את הגישה ברמת הפרויקט, התיקייה והארגון.

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

ההרשאות הנדרשות

כדי ליצור אשכול Connect, צריך את ההרשאות הבאות:

  • נותנים את ההרשאה ליצירת אשכול Connect במיקום שצוין: managedkafka.connectClusters.create

יכול להיות שתקבלו את ההרשאות האלה באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש אחרים.

חשבונות משתמשים נדרשים ב-ACL

כברירת מחדל, אשכולות של שירות מנוהל ל-Apache Kafka מאפשרים ל-Connect Cluster לגשת למשאבים אם לא הוגדרו רשימות ACL. כדי לעשות זאת, מגדירים את allow.everyone.if.no.acl.found כ-true, שזו הגדרת ברירת המחדל.

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

חשבון השירות של אשכול Connect שמשמש כחשבון הראשי ברשימות ה-ACL הוא בפורמט הבא: User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com.

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

/bin/kafka-acls.sh \
    --bootstrap-server BOOTSTRAP_ADDR \
    --command-config PATH_TO_CLIENT_PROPERTIES \
    --add \
    --allow-principal User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com \
    --operation READ --operation WRITE --topic *

/bin/kafka-acls.sh \
    --bootstrap-server BOOTSTRAP_ADDR \
    --command-config PATH_TO_CLIENT_PROPERTIES \
    --add \
    --allow-principal User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com \
    --operation READ --group *

למידע נוסף על הפקודות האלה, ראו הגדרת רשימות ACL של Apache Kafka לבקרת גישה מפורטת.

יצירת אשכול Connect בפרויקט אחר

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

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

כדי לתת את ההרשאות הנדרשות, מקצים את התפקיד סוכן שירות מנוהל של Kafka לסוכן השירות של אשכול Connect בפרויקט של אשכול Kafka. לדוגמה, אם יוצרים אשכול Kafka בפרויקט kafka-project ואשכול Connect בפרויקט connect-project, צריך לתת את התפקיד "סוכן שירות מנוהל של Kafka" ב-kafka-project לסוכן השירות שמשויך ל-connect-project.

כתובת האימייל של סוכן השירות היא בפורמט הבא: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, כאשר PROJECT_NUMBER הוא מספר הפרויקט. במאמר איך יוצרים סוכני שירות ומעניקים להם תפקידים מוסבר איך מקצים את התפקיד.

מאפיינים של אשכול Connect

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

שם האשכול ב-Connect

שם האשכול ב-Connect שאתם יוצרים. במאמר הנחיות למתן שמות למשאבים של שירות מנוהל ל-Apache Kafka מוסבר איך נותנים שמות לאשכולות Connect. אי אפשר לשנות את השם של אשכול.

אשכול Kafka ראשי

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

לאשכול אחד של שירות מנוהל ל-Apache Kafka יכולים להיות כמה אשכולות של Connect. אם בוחרים אשכול של שירות מנוהל ל-Apache Kafka בפרויקט אחר, צריך לוודא שמוגדרות ההרשאות המתאימות.

אי אפשר לעדכן לאשכול Kafka אחר אחרי שיוצרים את אשכול Connect.

היתרונות של מיקום משותף באזור מבחינת זמן האחזור ועלויות הרשת

כדי לצמצם את זמן האחזור ואת עלויות הרשת, מומלץ למקם את השירות המנוהל ל-Apache Kafka ואת אשכולות Connect באותו אזור. לדוגמה, נניח שאשכול השירות המנוהל ל-Apache Kafka נמצא ב-region-a, ואתם משתמשים במחבר sink כדי לכתוב נתונים מאשכול השירות המנוהל ל-Apache Kafka (מקור) לטבלה ב-BigQuery (sink) שנמצאת גם ב-region-a. אם פורסים את אשכול Connect ב-region-a, בחירת הפריסה הזו ממזערת את זמן האחזור של פעולת הכתיבה ב-BigQuery ומבטלת את עלויות ההעברה ברשת בין אזורים בין האשכול של השירות המנוהל ל-Apache Kafka לבין אשכול Connect.

שיקולי עלות וזמן אחזור במערכות מרובות

‫Kafka Connect משתמש במחברים כדי להעביר נתונים בין מערכות. צד אחד של המחבר תמיד יוצר אינטראקציה עם אשכול של שירות מנוהל ל-Apache Kafka. אשכול Kafka Connect אחד יכול להריץ כמה מחברים, שכל אחד מהם פועל כמקור (שליפת נתונים ממערכת) או ככיור (דחיפת נתונים למערכת).

אשכול Connect באותו אזור של אשכול השירות המנוהל ל-Apache Kafka נהנה מזמן אחזור קצר יותר בתקשורת ביניהם, אבל כל מחבר גם יוצר אינטראקציה עם מערכת אחרת, כמו טבלה ב-BigQuery או אשכול Kafka אחר. גם אם אשכול Connect ואשכול השירות המנוהל ל-Apache Kafka נמצאים באותו מיקום, יכול להיות שהמערכת השנייה נמצאת באזור אחר. זה מוביל לזמן אחזור ארוך יותר ולעלות גבוהה יותר. זמן האחזור הכולל של צינור עיבוד הנתונים תלוי במיקומים של כל שלוש המערכות: האשכול של השירות המנוהל ל-Apache Kafka, האשכול של Connect והמערכת של המקור או של ה-sink.

לדוגמה, אם אשכול השירות המנוהל ל-Apache Kafka נמצא ב-region-a, אשכול Connect נמצא ב-region-b, ואתם משתמשים בחיבור ל-Cloud Storage בשביל קטגוריה ב-region-c, תחויבו על שני דילוגים ברשת (מ-region-a ל-region-b ואז מ-region-b ל-region-c, או בסדר ההפוך, בהתאם לכיוון החיבור).

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

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

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

  • ‫vCPUs: מספר המעבדים הווירטואליים שהוקצו לאשכול Connect. הערך המינימלי הוא 3 מעבדי vCPU.

  • זיכרון: כמות הזיכרון שמוקצית לכל vCPU. צריך להקצות בין ‎1 GiB ל-‎8 GiB לכל vCPU. אחרי שיוצרים את האשכול, אפשר להגדיל או להקטין את נפח הזיכרון בטווח הזה.

    לדוגמה, אם יוצרים אשכול עם 6 מעבדי vCPU, הזיכרון המינימלי שאפשר להקצות לאשכול הוא ‎6 GiB (‎1 GiB לכל vCPU), והמקסימלי הוא ‎48 GiB (‎8 GiB לכל vCPU).

ל-vCPU ולהקצאת הזיכרון שמוקצים לכל worker באשכול Connect יש השפעה משמעותית על הביצועים, הקיבולת והעלות של האשכול. כאן מפורט איך vCPU וזיכרון משפיעים על אשכול Connect.

מספר מעבדי ה-vCPU

  • ‫Kafka Connect מחלק את העבודה של מחבר למשימות. כל משימה יכולה לעבד נתונים במקביל. ככל שיש יותר מעבדים וירטואליים, אפשר להריץ יותר משימות בו-זמנית, וכך להגדיל את התפוקה.

  • ככל שיש יותר מעבדי vCPU, העלויות של אשכול Connect עולות.

זיכרון

  • ‫Kafka Connect משתמש בזיכרון כדי לאחסן נתונים במאגר נתונים זמני (buffer) בזמן שהם עוברים בין מחברים לבין השירות המנוהל ל-Apache Kafka. זיכרון גדול יותר מאפשר מאגרים גדולים יותר. זיכרון גדול יכול לשפר את קצב העברת הנתונים, במיוחד בנפחים גדולים. מחברים שמטפלים בהודעות או ברשומות גדולות מאוד צריכים מספיק זיכרון כדי לעבד אותן בלי להיתקל בחריגות OutOfMemoryError.

  • ככל שמוסיפים יותר זיכרון, העלות של אשכול Connect עולה.

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

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

רשת משנה של Worker (ראשית)

רשת המשנה של ה-worker, שנקראת גם רשת המשנה הראשית, מחברת את רשת ה-VPC לאשכול Connect. רשת המשנה הזו מאפשרת ל-workers של האשכול להגיע לנקודות הקצה של המקורות והכיורים ברשת הצרכן, כמו אשכולות של שירות מנוהל ל-Apache Kafka או אשכולות של Kafka באירוח עצמי.

הנה כמה דרישות להגדרת רשת המשנה של ה-worker:

  • חובה לציין את רשת המשנה של ה-worker.

  • רשת המשנה צריכה להיות באותו אזור שבו נמצא אשכול Connect.

  • רשת המשנה צריכה להיות באותו VPC הורה כמו אחת מרשתות המשנה המחוברות ברשימה של אשכול Kafka הראשי.

  • טווח ה-CIDR של רשת המשנה חייב להיות בגודל מינימלי של ‎/22 (1,024 כתובות).

ל-workers של האשכול מוקצות כתובות IP בתת-הרשת של ה-workers, באמצעות ממשק Private Service Connect. ה-Workers יכולים להגיע לכל יעד ברשת שאפשר לגשת אליו מרשת ה-VPC של רשת המשנה, בכפוף לדרישות הבאות:

  • נקודת הקצה לא יכולה להיות בטווח ה-CIDR של 172.16.0.0/14. הטווח הזה שמור לשימוש פנימי של השירות המנוהל ל-Apache Kafka Connect.
  • כללי חומת האש צריכים לאפשר את התנועה. הגדרת אבטחה לחיבורי רשת
  • לתעבורת נתונים באינטרנט, צריך להגדיר Cloud NAT. לדוגמה, נדרש Cloud NAT למחבר MirrorMaker כדי לשכפל נתונים מאשכול Kafka שנגיש דרך האינטרנט.
  • כדי לגשת לנקודות קצה של Private Service Connect שנמצאות ברשת VPC שונה מרשת ה-VPC של רשת המשנה של ה-worker, צריך לוודא שאתם משתמשים בהגדרת צרכן נתמכת (לדוגמה, NCC). מידע נוסף זמין במאמר מידע על גישה לשירותים שפורסמו דרך נקודות קצה.

דומיינים של DNS שניתנים לפענוח

דומיינים של DNS שניתנים לפתרון, שנקראים גם שמות דומיינים של DNS, מאפשרים לכתובות DNS ברשת ה-VPC של הצרכן להיות זמינות ל-VPC של הדייר. כך אשכול Connect יכול לפתור שמות DNS לכתובות IP, מה שמאפשר תקשורת עם שירותים אחרים, כולל אשכולות Kafka אחרים למחברי MirrorMaker.

בדומיינים של DNS שאפשר לפתור, אתם יכולים לבחור אשכול של שירות מנוהל ל-Apache Kafka. לא צריך להגדיר את שם הדומיין ב-DNS לאשכול הראשי של השירות המנוהל ל-Apache Kafka, כי כתובת ה-bootstrap שלו נכללת באופן אוטומטי ברשימת הדומיינים ב-DNS שאפשר לפתור.

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

משאבים של Secret Manager

חלק מהמחברים דורשים נתונים רגישים, כמו סיסמאות, כחלק מההגדרה שלהם. כדי לנהל את סוג הנתונים הזה בצורה מאובטחת, אתם יכולים לאחסן את הנתונים ב-Secret Manager ולתת לאשכול Connect גישה לסוד.

כדי להשתמש בסודות של Secret Manager עם Kafka Connect:

  1. מקצים לחשבון השירות של Managed Kafka את התפקידים הבאים ב-IAM. התפקידים האלה מאפשרים לאשכול Connect להציג סודות ולגשת למטענים הייעודיים שלהם.

    • צפייה ב-Secret Manager ‏ (roles/secretmanager.viewer)
    • ממשק גישה לסודות ב-Secret Manager‏ (roles/secretmanager.secretAccessor)

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

  3. כשיוצרים או מעדכנים את אשכול Connect, מציינים את הסודות שלאשכול יש גישה אליהם. אפשר לציין עד 32 סודות לכל אשכול Connect.

הסודות מותקנים כקבצים ב-workers של האשכול. למחברים יש הרשאת קריאה בלבד לקבצים האלה. כשיוצרים מחבר, מאפייני ההגדרה של המחבר יכולים להפנות לסודות.

  • כדי להפנות לנתיב של קובץ סודי, משתמשים בפורמט הבא:

    /var/secrets/PROJECT_NAME-SECRET_NAME-SECRET_VERSION

    לדוגמה: ssl.truststore.location=/var/secrets/project1-truststore-1

  • כדי להשתמש ב-value של סוד כערך הגדרה (לדוגמה, סיסמה), משתמשים בפורמט הבא:

    ${directory:/var/secrets:PROJECT_NAME-SECRET_NAME-SECRET_VERSION}

    לדוגמה: password=${directory:/var/secrets:project1-database_password-3}

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

  • ‫PROJECT_NAME: שם הפרויקט ב- Google Cloud .
  • ‫SECRET_NAME: שם הסוד.
  • ‫SECRET_VERSION: גרסת הסוד.

תוויות

תוויות הן צמדי מפתח/ערך שעוזרים לכם לארגן ולזהות את המשאבים. הם עוזרים לארגן את אשכולות Connect. אפשר לצרף תווית לכל אשכול Connect, ואז לסנן את המשאבים לפי התוויות שלהם. דוגמאות לתוויות: environment:prod, application:web-app.

יצירת אשכול Connect

לפני שיוצרים אשכול, כדאי לעיין במסמכי התיעוד של מאפייני אשכולות של Connect.

יצירת אשכול Connect נמשכת בין 20 ל-30 דקות.

המסוף

  1. נכנסים לדף Connect Clusters במסוף Google Cloud .

    מעבר אל Connect Clusters

  2. לוחצים על Create.

  3. בשדה Connect cluster name, נותנים שם לאשכול Connect. מידע נוסף זמין במאמר הנחיות למתן שמות למשאבים של שירות מנוהל ל-Apache Kafka.

  4. ברשימה Primary Kafka cluster, בוחרים אשכול של השירות המנוהל ל-Apache Kafka. מידע נוסף זמין במאמר אשכול Kafka ראשי.

  5. ברשימה Region, בוחרים מיקום לאשכול Connect. למידע נוסף על בחירת מיקום, ראו אשכול Kafka ראשי.

  6. בקטע Capacity configuration, מזינים ערכים בשדות הבאים או משאירים את ערכי ברירת המחדל.

    • בשדה vCPUs, מזינים את מספר המעבדים הווירטואליים של האשכול.

    • בשדה Memory, מזינים את נפח הזיכרון לכל מעבד (CPU), ב-GiB. הערך לא יכול להיות גבוה מ-‎8GiB לכל מעבד (CPU).

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

  7. בקטע Network configuration, בוחרים רשת VPC מהרשימה Network או משאירים את ערך ברירת המחדל. הרשימה הזו מאוכלסת כשבוחרים את אשכול Kafka הראשי.

  8. בקטע Worker subnet, בוחרים רשת משנה מהרשימה Subnet או משאירים את ערך ברירת המחדל. מידע נוסף זמין במאמר רשת משנה של Worker. השדה Subnet URI path מתמלא באופן אוטומטי כשבוחרים את רשת המשנה.

  9. אופציונלי: מוסיפים דומיין DNS שניתן לפתור. דומיין ה-DNS של אשכול Kafka הראשי מתווסף באופן אוטומטי כדומיין DNS שניתן לפתור. כדי לציין דומיינים נוספים של DNS:

    1. מרחיבים את הקטע דומיינים של DNS שניתנים לפתרון.

    2. לוחצים על הוספת דומיין DNS.

    3. כדי להוסיף את דומיין ה-DNS של אשכול קיים של שירות מנוהל ל-Apache Kafka, בוחרים את האשכול מהרשימה Kafka cluster. אחרת, מזינים את דומיין ה-DNS בשדה DNS domain.

    4. לוחצים על סיום.

  10. אופציונלי: כדי להוסיף משאבים של Secret Manager:

    1. מרחיבים את הקטע Secret manager resources.

    2. לוחצים על Add secret resource.

    3. ברשימה Secret, בוחרים סוד.

    4. ברשימה Secret version, בוחרים גרסה של הסוד.

    5. לוחצים על סיום.

  11. אופציונלי: מוסיפים תוויות כדי לארגן את הפרויקט. כדי להוסיף תווית:

    1. מרחיבים את הקטע Labels.

    2. לוחצים על הוספת תווית.

    3. בשדה Key, מזינים את המפתח של התווית.

    4. בשדה ערך, מזינים את הערך של התווית.

  12. לוחצים על יצירה.

gcloud

  1. מפעילים את Cloud Shell במסוף Google Cloud .

    הפעלת Cloud Shell

    בחלק התחתון של מסוף Google Cloud יתחיל סשן של Cloud Shell ותופיע הודעה של שורת הפקודה. Cloud Shell היא סביבת מעטפת שבה ה-CLI של Google Cloud מותקן ומוגדרים ערכים לפרויקט הקיים. הסשן יופעל תוך כמה שניות.

  2. מריצים את הפקודה gcloud managed-kafka connect-clusters create:

    gcloud managed-kafka connect-clusters create CONNECT_CLUSTER_ID \
    --location=LOCATION \
    --cpu=CPU \
    --memory=MEMORY \
    --primary-subnet=WORKER_SUBNET \
    --kafka-cluster=KAFKA_CLUSTER \
    [--project=PROJECT_ID] \
    [--secret=SECRET] \
    [--dns-name=DNS_DOMAIN_NAME] \
    [--config-file=CONFIG_FILE] \
    [--labels=LABELS]
    [--async]

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

    • ‫CONNECT_CLUSTER_ID: השם או המזהה של אשכול Connect. לקבלת הנחיות למתן שמות לאשכול Connect, ראו הנחיות למתן שמות למשאבים של שירות מנוהל ל-Apache Kafka. אי אפשר לשנות את השם של אשכול Connect.

    • ‫LOCATION: המיקום שבו יוצרים את אשכול Connect. הכתובת צריכה להיות באזור נתמך של Google Cloud. אי אפשר לשנות את המיקום של אשכול Connect אחרי שיוצרים אותו. רשימת המיקומים הזמינים לשירות מנוהל ל-Apache Kafka מידע נוסף על המלצות למיקום זמין במאמר אשכול Kafka ראשי.

    • ‫CPU: מספר ה-vCPU של אשכול Connect. הערך המינימלי הוא 3 מעבדי vCPU. מידע נוסף זמין במאמר מספר מעבדי ה-VCPU.

    • ‫MEMORY: כמות הזיכרון של אשכול Connect. אפשר להשתמש ביחידות "MB",‏ "MiB",‏ "GB",‏ "GiB",‏ "TB" או "TiB". לדוגמה, "3GiB". צריך להקצות בין ‎1GiB ל-‎8GiB לכל vCPU. למידע נוסף על זיכרון

    • ‫WORKER_SUBNET: רשת המשנה של ה-worker באשכול Connect.

      הפורמט של רשת המשנה הוא projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.

      רשת המשנה של ה-worker צריכה להיות באותו אזור שבו נמצא אשכול Connect.

    • ‫PROJECT_ID: (אופציונלי) המזהה של הפרויקט ב-Google Cloud . אם לא מציינים פרויקט, המערכת משתמשת בפרויקט הנוכחי.

    • ‫KAFKA_CLUSTER: המזהה או השם המלא של אשכול שירות מנוהל ל-Apache Kafka הראשי שמשויך לאשכול Connect. אשכול Kafka הפורמט של אשכול Kafka הוא projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID.

      אי אפשר לעדכן לאשכול Kafka אחר אחרי שיוצרים את אשכול Connect.

    • ‫SECRET: (אופציונלי) סודות לטעינה ב-workers. צריך לספק את הגרסאות המדויקות של הסודות מ-Secret Manager, כי אין תמיכה בכינויים. אפשר לטעון עד 32 סודות לאשכול אחד. פורמט: projects/PROJECT_ID/secrets/SECRET_NAME/versions/VERSION_ID

    • ‫DNS_DOMAIN_NAME: (אופציונלי) שמות דומיינים של DNS מתת-הרשת שיהיו גלויים לאשכול Connect. לאשכול Connect יש גישה למשאבים באמצעות שמות דומיינים, במקום להסתמך על כתובות IP. מידע נוסף זמין במאמר קישור בין רשתות שכנות (peering) ב-DNS.

    • ‫LABELS: (אופציונלי) תוויות לשיוך לאשכול. מידע נוסף על הפורמט של תוויות רשימה של צמדי מפתח/ערך של תוויות שרוצים להוסיף. המפתחות צריכים להתחיל באות קטנה ולהכיל רק אותיות קטנות, ספרות, מקפים (-) וקווים תחתונים (_). הערכים יכולים להכיל רק מקפים (-), קווים תחתונים (_), אותיות קטנות ומספרים.

    • ‫CONFIG_FILE: (אופציונלי) הנתיב לקובץ ה-JSON או ה-YAML שמכיל את ההגדרות שמחליפות את ברירות המחדל של האשכול או המחבר. הקובץ הזה תומך גם ב-JSON או ב-YAML בתוך השורה.

    • ‫--async: (אופציונלי) חזרה מיידית, בלי לחכות שהפעולה תסתיים. באמצעות הדגל --async, אתם יכולים להמשיך במשימות אחרות בזמן שיצירת האשכול מתרחשת ברקע. אם לא משתמשים בדגל, המערכת ממתינה עד שהפעולה תסתיים לפני שהיא מחזירה תשובה. צריך להמתין עד שהאשכול יעודכן באופן מלא לפני שאפשר יהיה להמשיך במשימות אחרות.

    התגובה אמורה להיראות כך:

    Create request issued for: [sample-connectcluster]
Check operation [projects/test-project/locations/us-east1/operations/operation-1753590328249-63ae19098cc06-64300a0a-06512d02] for status.

    שומרים את OPERATION_ID כדי לעקוב אחרי ההתקדמות. לדוגמה, הערך כאן הוא operation-1753590328249-63ae19098cc06-64300a0a-06512d02.

Terraform

אתם יכולים להשתמש במשאב של Terraform כדי ליצור אשכול Connect.

resource "google_managed_kafka_connect_cluster" "default" {
  provider           = google-beta
  project            = data.google_project.default.project_id
  connect_cluster_id = "my-connect-cluster-id"
  location           = "us-central1"
  kafka_cluster      = google_managed_kafka_cluster.default.id
  capacity_config {
    vcpu_count   = 12
    memory_bytes = 12884901888 # 12 GiB
  }
  gcp_config {
    access_config {
      network_configs {
        primary_subnet = google_compute_subnetwork.default.id
      }
    }
  }
}

כדי ללמוד איך להחיל הגדרות ב-Terraform או להסיר אותן, ראו פקודות בסיסיות ב-Terraform.

המשך

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

כדי לבצע אימות בשירות המנוהל ל-Apache Kafka, צריך להגדיר את Application Default Credentials‏(ADC). מידע נוסף זמין במאמר הגדרת ADC לסביבת פיתוח מקומית.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
)

func createConnectCluster(w io.Writer, projectID, region, clusterID, kafkaCluster string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-connect-cluster"
	// kafkaCluster := "projects/my-project-id/locations/us-central1/clusters/my-kafka-cluster"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	locationPath := fmt.Sprintf("projects/%s/locations/%s", projectID, region)
	clusterPath := fmt.Sprintf("%s/connectClusters/%s", locationPath, clusterID)

	// Capacity configuration with 12 vCPU and 12 GiB RAM
	capacityConfig := &managedkafkapb.CapacityConfig{
		VcpuCount:   12,
		MemoryBytes: 12884901888, // 12 GiB in bytes
	}

	// Optionally, you can also specify accessible subnets and resolvable DNS
	// domains as part of your network configuration. For example:
	// networkConfigs := []*managedkafkapb.ConnectNetworkConfig{
	// 	{
	// 		PrimarySubnet:      primarySubnet,
	// 		AdditionalSubnets:  []string{"subnet-1", "subnet-2"},
	// 		DnsDomainNames:     []string{"domain-1", "domain-2"},
	// 	},
	// }

	connectCluster := &managedkafkapb.ConnectCluster{
		Name:           clusterPath,
		KafkaCluster:   kafkaCluster,
		CapacityConfig: capacityConfig,
	}

	req := &managedkafkapb.CreateConnectClusterRequest{
		Parent:           locationPath,
		ConnectClusterId: clusterID,
		ConnectCluster:   connectCluster,
	}
	op, err := client.CreateConnectCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateConnectCluster got err: %w", err)
	}
	// The duration of this operation can vary considerably, typically taking 5-15 minutes.
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Created connect cluster: %s\n", resp.Name)
	return nil
}

Java

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Java בקטע התקנת ספריות הלקוח. מידע נוסף מופיע ב מאמרי העזרה של Managed Service for Apache Kafka Java API.

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


import com.google.api.gax.longrunning.OperationFuture;
import com.google.api.gax.longrunning.OperationSnapshot;
import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
import com.google.api.gax.retrying.RetrySettings;
import com.google.api.gax.retrying.RetryingFuture;
import com.google.api.gax.retrying.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.ConnectAccessConfig;
import com.google.cloud.managedkafka.v1.ConnectCluster;
import com.google.cloud.managedkafka.v1.ConnectGcpConfig;
import com.google.cloud.managedkafka.v1.ConnectNetworkConfig;
import com.google.cloud.managedkafka.v1.CreateConnectClusterRequest;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectSettings;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class CreateConnectCluster {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String clusterId = "my-connect-cluster";
    String subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
    String kafkaCluster = "my-kafka-cluster"; // The Kafka cluster to connect to
    int cpu = 12;
    long memoryBytes = 12884901888L; // 12 GiB
    createConnectCluster(projectId, region, clusterId, subnet, kafkaCluster, cpu, memoryBytes);
  }

  public static void createConnectCluster(
      String projectId,
      String region,
      String clusterId,
      String subnet,
      String kafkaCluster,
      int cpu,
      long memoryBytes)
      throws Exception {
    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setVcpuCount(cpu)
        .setMemoryBytes(memoryBytes).build();
    ConnectNetworkConfig networkConfig = ConnectNetworkConfig.newBuilder()
        .setPrimarySubnet(subnet)
        .build();
    // Optionally, you can also specify additional accessible subnets and resolvable
    // DNS domains as part of your network configuration. For example:
    // .addAllAdditionalSubnets(List.of("subnet-1", "subnet-2"))
    // .addAllDnsDomainNames(List.of("dns-1", "dns-2"))
    ConnectGcpConfig gcpConfig = ConnectGcpConfig.newBuilder()
        .setAccessConfig(ConnectAccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
        .build();
    ConnectCluster connectCluster = ConnectCluster.newBuilder()
        .setCapacityConfig(capacityConfig)
        .setGcpConfig(gcpConfig)
        .setKafkaCluster(kafkaCluster)
        .build();

    // Create the settings to configure the timeout for polling operations
    ManagedKafkaConnectSettings.Builder settingsBuilder = ManagedKafkaConnectSettings.newBuilder();
    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
        RetrySettings.newBuilder()
            .setTotalTimeoutDuration(Duration.ofHours(1L))
            .build());
    settingsBuilder.createConnectClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient
        .create(settingsBuilder.build())) {
      CreateConnectClusterRequest request = CreateConnectClusterRequest.newBuilder()
          .setParent(LocationName.of(projectId, region).toString())
          .setConnectClusterId(clusterId)
          .setConnectCluster(connectCluster)
          .build();

      // The duration of this operation can vary considerably, typically taking
      // between 10-30 minutes.
      OperationFuture<ConnectCluster, OperationMetadata> future = managedKafkaConnectClient
          .createConnectClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf(
          "Connect cluster creation started. Operation name: %s\nDone: %s\nMetadata: %s\n",
          operation.getName(), operation.isDone(), future.getMetadata().get().toString());

      while (!future.isDone()) {
        // The pollingFuture gives us the most recent status of the operation
        RetryingFuture<OperationSnapshot> pollingFuture = future.getPollingFuture();
        OperationSnapshot currentOp = pollingFuture.getAttemptResult().get();
        System.out.printf("Polling Operation:\nName: %s\n Done: %s\n",
            currentOp.getName(),
            currentOp.isDone());
      }

      // NOTE: future.get() blocks completion until the operation is complete (isDone
      // = True)
      ConnectCluster response = future.get();
      System.out.printf("Created connect cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaConnectClient.createConnectCluster got err: %s\n", 
          e.getMessage());
      throw e;
    }
  }
}

Python

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

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

from google.api_core.exceptions import GoogleAPICallError
from google.cloud import managedkafka_v1
from google.cloud.managedkafka_v1.services.managed_kafka_connect import ManagedKafkaConnectClient
from google.cloud.managedkafka_v1.types import ConnectCluster, CreateConnectClusterRequest, ConnectNetworkConfig

# TODO(developer): Update with your values.
# project_id = "my-project-id"
# region = "us-central1"
# connect_cluster_id = "my-connect-cluster"
# kafka_cluster_id = "my-kafka-cluster"
# primary_subnet = "projects/my-project-id/regions/us-central1/subnetworks/default"
# cpu = 12
# memory_bytes = 12884901888  # 12 GiB

connect_client = ManagedKafkaConnectClient()
kafka_client = managedkafka_v1.ManagedKafkaClient()

parent = connect_client.common_location_path(project_id, region)
kafka_cluster_path = kafka_client.cluster_path(project_id, region, kafka_cluster_id)

connect_cluster = ConnectCluster()
connect_cluster.name = connect_client.connect_cluster_path(project_id, region, connect_cluster_id)
connect_cluster.kafka_cluster = kafka_cluster_path
connect_cluster.capacity_config.vcpu_count = cpu
connect_cluster.capacity_config.memory_bytes = memory_bytes
connect_cluster.gcp_config.access_config.network_configs = [ConnectNetworkConfig(primary_subnet=primary_subnet)]
# Optionally, you can also specify accessible subnets and resolvable DNS domains as part of your network configuration.
# For example:
# connect_cluster.gcp_config.access_config.network_configs = [
#     ConnectNetworkConfig(
#         primary_subnet=primary_subnet,
#         additional_subnets=additional_subnets,
#         dns_domain_names=dns_domain_names,
#     )
# ]

request = CreateConnectClusterRequest(
    parent=parent,
    connect_cluster_id=connect_cluster_id,
    connect_cluster=connect_cluster,
)

try:
    operation = connect_client.create_connect_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    # Creating a Connect cluster can take 10-40 minutes.
    response = operation.result(timeout=3000)
    print("Created Connect cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e}")

מעקב אחרי פעולת יצירת האשכול

אפשר להריץ את הפקודה הבאה רק אם הרצתם את ה-CLI של gcloud כדי ליצור את אשכול Connect.

  • בדרך כלל, יצירת אשכול Connect נמשכת 20-30 דקות. כדי לעקוב אחרי ההתקדמות של יצירת האשכול, הפקודה gcloud managed-kafka connect-clusters create משתמשת בפעולה ממושכת (LRO), שאפשר לעקוב אחריה באמצעות הפקודה הבאה:

    gcloud managed-kafka operations describe OPERATION_ID \
    --location=LOCATION

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

    • ‫OPERATION_ID עם הערך של מזהה הפעולה מהקטע הקודם.
    • ‫LOCATION עם ערך המיקום מהקטע הקודם.

מה השלב הבא?

‫Apache Kafka®‎ הוא סימן מסחרי רשום של The Apache Software Foundation או של השותפים העצמאיים שלה בארצות הברית או במדינות אחרות.