יצירת נושא לייבוא ב-Amazon Managed Streaming for Apache Kafka

נושא ייבוא של Amazon Managed Streaming for Apache Kafka‏ (Amazon MSK) מאפשר לכם להטמיע נתונים מ-Amazon MSK כמקור חיצוני ב-Pub/Sub באופן רציף. אחר כך אפשר להעביר את הנתונים בסטרימינג לכל אחד מהיעדים שנתמכים ב-Pub/Sub.

במאמר הזה מוסבר איך ליצור ולנהל נושאים לייבוא של Amazon MSK. כדי ליצור נושא רגיל, ראו יצירת נושא רגיל.

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

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

תפקידים והרשאות נדרשים

כדי לקבל את ההרשאות שנדרשות ליצירה ולניהול של נושאי ייבוא של Amazon MSK, צריך לבקש מהאדמין להקצות לכם ב-IAM את התפקיד עריכה ב-Pub/Sub (roles/pubsub.editor) בנושא או בפרויקט. כדי לקרוא הסבר על מתן תפקידים, ראו איך מנהלים את הגישה ברמת הפרויקט, התיקייה והארגון.

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

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

כדי ליצור ולנהל נושאי ייבוא של Amazon MSK, צריך את ההרשאות הבאות:

  • יצירת נושא ייבוא: pubsub.topics.create
  • מחיקת נושא ייבוא: pubsub.topics.delete
  • קבלת נושא ייבוא: pubsub.topics.get
  • רשימה של נושא ייבוא: pubsub.topics.list
  • פרסום בנושא ייבוא: pubsub.topics.publish ו-pubsub.serviceAgent
  • עדכון נושא ייבוא: pubsub.topics.update
  • קבלת מדיניות ה-IAM של נושא הייבוא: pubsub.topics.getIamPolicy
  • מגדירים את מדיניות ה-IAM לנושא ייבוא: pubsub.topics.setIamPolicy

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

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

הגדרת זהות מאוחדת כדי לגשת ל-Amazon MSK

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

יצירת חשבון שירות ב Google Cloud

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

בנושאי ייבוא של Amazon MSK, ‏ Pub/Sub משתמש בחשבון השירות כזהות כדי לגשת למשאבים מ-AWS.

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

תיעוד המזהה הייחודי של חשבון השירות

כדי להגדיר תפקיד במסוף AWS, צריך מזהה ייחודי של חשבון שירות.

  1. נכנסים לדף הפרטים Service account במסוף Google Cloud .

    כניסה לדף Service accounts

  2. לוחצים על חשבון השירות שיצרתם או על החשבון שאתם מתכננים להשתמש בו.

  3. בדף Service account details, רושמים את המזהה הייחודי.

    תצטרכו את המזהה כחלק מתהליך העבודה להגדרת תפקיד במסוף AWS.

הוספת התפקיד "יצירת אסימונים בחשבון שירות" לחשבון השירות של Pub/Sub

התפקיד 'יצירת אסימונים בחשבון שירות' (roles/iam.serviceAccountTokenCreator) מאפשר לחשבונות משתמשים ליצור פרטי כניסה לטווח קצר לחשבון שירות. האסימונים או פרטי הכניסה האלה משמשים להתחזות לחשבון השירות.

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

אפשר גם להוסיף את התפקיד 'פרסום הודעות ב-Pub/Sub' (roles/pubsub.publisher) במהלך התהליך הזה. מידע נוסף על התפקיד ועל הסיבה להוספתו זמין במאמר הוספת התפקיד 'פרסום הודעות ב-Pub/Sub' לחשבון השירות של Pub/Sub.

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

    כניסה לדף IAM

  2. מסמנים את התיבה Include Google-provided role grants.

  3. מחפשים את חשבון השירות בפורמט service-{PROJECT_NUMBER}@gcp-sa-pubsub..

  4. לוחצים על הלחצן Edit Principal (עריכת חשבון המשתמש) בחשבון השירות הזה.

  5. אם צריך, לוחצים על Add another role.

  6. מחפשים את התפקיד 'יצירת אסימונים בחשבון שירות' (roles/iam.serviceAccountTokenCreator) ולוחצים עליו.

  7. לוחצים על Save.

יצירת מדיניות ב-AWS

צריך מדיניות ב-AWS כדי לאפשר ל-Pub/Sub לבצע אימות ב-AWS, כדי ש-Pub/Sub יוכל להטמיע נתונים מ-Amazon MSK.

כדי ליצור מדיניות ב-AWS:

  1. נכנסים ל-AWS Management Console ופותחים את מסוף IAM.

  2. בחלונית הניווט של המסוף IAM, לוחצים על ניהול הרשאות גישה ‏ > Policies.

  3. לוחצים על יצירת מדיניות.

  4. בקטע Click a service (בחירת שירות), לוחצים על MSK.

  5. בקטע Action allowed (הפעולה מותרת), לוחצים על Read (קריאה) > GetBootstrapBrokers.

    הפעולה הזו מעניקה הרשאה לקבל את ברוקרי האתחול ש-Pub/Sub משתמש בהם כדי להתחבר לאשכול MSK.

  6. לוחצים על הוספת הרשאות נוספות.

  7. בקטע Select a service (בחירת שירות), לוחצים על Apache Kafka APIs for MSK (ממשקי API של Apache Kafka ל-MSK).

  8. בקטע Action allowed, בוחרים באפשרויות הבאות:

    • ‫List > DescribeTopic

      הפעולה הזו נותנת הרשאה לנושא ההטמעה של Pub/Sub לקבל פרטים על נושא Amazon MSK Kafka.

    • קריאה > ReadData

      הפעולה הזו מעניקה הרשאה לקרוא נתונים מנושא Amazon MSK Kafka.

    • Write‏ > Connect

      הפעולה הזו נותנת הרשאה להתחבר לאשכול Amazon MSK Kafka ולאמת אותו.

  9. בקטע Resources, מציינים את cluster ARN (אם רוצים להגביל את המדיניות לאשכולות ספציפיים, וזה מומלץ).

  10. לוחצים על הוספת הרשאות נוספות.

  11. בקטע Select a service (בחירת שירות), לוחצים על STS.

  12. בקטע Action allowed, לוחצים על Write ואז על AssumeRoleWithWebIdentity.

    הפעולה הזו מעניקה הרשאה לקבל קבוצה של פרטי כניסה זמניים לאבטחה עבור Pub/Sub כדי לבצע אימות ל-Amazon MSK באמצעות איחוד שירותי אימות הזהות.

  13. לוחצים על הבא.

  14. מזינים שם ותיאור למדיניות.

  15. לוחצים על יצירת מדיניות.

יצירת תפקיד ב-AWS באמצעות מדיניות אמון בהתאמה אישית

כדי ש-Pub/Sub יוכל לבצע אימות ב-AWS ולהטמיע נתונים מ-Amazon MSK, צריך ליצור תפקיד ב-AWS.

  1. נכנסים ל-AWS Management Console ופותחים את IAM console.

  2. בחלונית הניווט של המסוף, בקטע IAM, לוחצים על Roles (תפקידים).

  3. לוחצים על יצירת תפקיד.

  4. בקטע Select trusted entity, לוחצים על Custom trust policy.

  5. בקטע Custom trust policy, מזינים או מדביקים את הטקסט הבא:

    {
  "Version": "2012-10-17",
  "Statement": [
    {
     "Effect": "Allow",
     "Principal": {
        "Federated": "accounts.google.com"
     },
     "Action": "sts:AssumeRoleWithWebIdentity",
     "Condition": {
         "StringEquals": {
           "accounts.google.com:sub": "<SERVICE_ACCOUNT_UNIQUE_ID>"
         }
      }
    }
  ]
}

    מחליפים את <SERVICE_ACCOUNT_UNIQUE_ID> במזהה הייחודי של חשבון השירות שרשמתם בשלב רשימת המזהה הייחודי של חשבון השירות.

  6. לוחצים על הבא.

  7. בקטע Add permissions, מחפשים את המדיניות המותאמת אישית שיצרתם ולוחצים עליה.

  8. לוחצים על הבא.

  9. מזינים שם ותיאור לתפקיד.

  10. לוחצים על יצירת תפקיד.

הוספת התפקיד 'פרסום הודעות ב-Pub/Sub' לחשבון המשתמש ב-Pub/Sub

כדי להפעיל פרסום, צריך להקצות תפקיד של פרסום לחשבון השירות של Pub/Sub, כדי ש-Pub/Sub יוכל לפרסם בנושא הייבוא של Amazon MSK.

הוספת התפקיד "סוכן שירות Pub/Sub" לחשבון השירות של Pub/Sub

כדי לאפשר ל-Pub/Sub להשתמש במכסת הפרסום של פרויקט נושא הייבוא, לסוכן השירות של Pub/Sub נדרשת ההרשאה serviceusage.services.use בפרויקט של נושא הייבוא.

כדי לתת את ההרשאה הזו, מומלץ להוסיף את תפקיד סוכן השירות של Pub/Sub לחשבון השירות של Pub/Sub.

אם לחשבון השירות של Pub/Sub אין את התפקיד של סוכן שירות Pub/Sub, אפשר להקצות אותו באופן הבא:

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

    כניסה לדף IAM

  2. מסמנים את התיבה Include Google-provided role grants.

  3. מחפשים את חשבון השירות בפורמט service-{PROJECT_NUMBER}@gcp-sa-pubsub..

  4. לוחצים על הלחצן Edit principal (עריכת חשבון המשתמש) בשורה של חשבון השירות הרלוונטי.

  5. אם צריך, לוחצים על Add another role.

  6. מחפשים את התפקיד של סוכן שירות Pub/Sub (roles/pubsub.serviceAgent) ולוחצים עליו.

  7. לוחצים על Save.

הפעלת פרסום מכל הנושאים

כדאי להשתמש בשיטה הזו אם לא יצרתם נושאים לייבוא ב-Amazon MSK.

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

    כניסה לדף IAM

  2. מסמנים את התיבה Include Google-provided role grants.

  3. מחפשים את חשבון השירות בפורמט service-{PROJECT_NUMBER}@gcp-sa-pubsub..

  4. לוחצים על הלחצן Edit principal של חשבון השירות הזה.

  5. אם צריך, לוחצים על הוספת תפקיד נוסף.

  6. מחפשים את התפקיד 'פרסום הודעות ב-Pub/Sub' (roles/pubsub.publisher) ולוחצים עליו.

  7. לוחצים על Save.

הפעלת פרסום מנושא יחיד

משתמשים בשיטה הזו רק אם נושא הייבוא של Amazon MSK כבר קיים.

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

    הפעלת Cloud Shell

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

  2. מריצים את הפקודה gcloud pubsub topics add-iam-policy-binding:

    gcloud pubsub topics add-iam-policy-binding TOPIC_ID \
   --member="serviceAccount:service-PROJECT_NUMBER@gcp-sa-pubsub." \
   --role="roles/pubsub.publisher"

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

    • ‫TOPIC_ID: מזהה הנושא של נושא הייבוא של Amazon MSK.

    • ‫PROJECT_NUMBER: מספר הפרויקט. כדי לראות את מספר הפרויקט, קראו את המאמר זיהוי פרויקטים.

הוספת התפקיד 'משתמש בחשבון שירות' לחשבון השירות

התפקיד 'משתמש בחשבון השירות' (roles/iam.serviceAccountUser) כולל את ההרשאה iam.serviceAccounts.actAs שמאפשרת לישות מורשית לצרף חשבון שירות להגדרות הטמעת הנתונים של נושא הייבוא של Amazon MSK ולהשתמש בחשבון השירות הזה לצורך זהות מאוחדת.

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

    כניסה לדף IAM

  2. לוחצים על הלחצן Edit principal (עריכת חשבון המשתמש) בשורה של חשבון המשתמש שמוציא את הקריאות ליצירה או לעדכון של נושא.

  3. אם צריך, לוחצים על הוספת תפקיד נוסף.

  4. מחפשים את התפקיד 'משתמש בחשבון שירות' (roles/iam.serviceAccountUser) ולוחצים עליו.

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

שימוש בנושאי ייבוא של Amazon MSK

אתם יכולים ליצור נושא ייבוא חדש או לערוך נושא קיים.

לתשומת ליבכם

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

  • אם אתם צריכים ליצור מחדש את נושא Kafka של נושא ייבוא קיים עם אותו שם, אתם לא יכולים פשוט למחוק את נושא Kafka וליצור אותו מחדש. הפעולה הזו עלולה לגרום לביטול התוקף של ניהול ההיסטים ב-Pub/Sub, מה שעלול להוביל לאובדן נתונים. כדי לפתור את הבעיה, מבצעים את השלבים הבאים:

    • מוחקים את נושא הייבוא ב-Pub/Sub.
    • מוחקים את נושא Kafka.
    • יוצרים את הנושא ב-Kafka.
    • יוצרים את נושא הייבוא ב-Pub/Sub.

  • הנתונים מנושא של Amazon MSK Kafka תמיד נקראים מההיסט המוקדם ביותר.

יצירת נושאי ייבוא של Amazon MSK

מידע נוסף על מאפיינים שמשויכים לנושא

ודאו שהשלמתם את הפעולות הבאות:

כדי ליצור נושא לייבוא מ-Amazon MSK:

המסוף

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

    מעבר אל "נושאים"

  2. לוחצים על יצירת נושא.

  3. בשדה Topic ID (מזהה הנושא), מזינים מזהה לנושא הייבוא של Amazon MSK. מידע נוסף על מתן שמות לנושאים זמין בהנחיות למתן שמות.

  4. בוחרים באפשרות הוספת מינוי ברירת מחדל.

  5. לוחצים על הפעלת הטמעה.

  6. בוחרים באפשרות Amazon MSK כמקור ההטמעה.

  7. מזינים את הפרטים הבאים:

    • ‫Cluster ARN: ה-ARN של Amazon MSK שאתם מטמיעים ב-Pub/Sub. הפורמט של ה-ARN הוא: arn:aws:kafka:${Region}:${Account}:cluster/${ClusterName}/${ClusterId}.
    • ‫Topic: השם של נושא Amazon MSK Kafka שמתבצעת ממנו העברה ל-Pub/Sub.
    • ‫AWS Role ARN: ה-ARN של התפקיד ב-AWS. הפורמט של ה-ARN של התפקיד הוא arn:aws:iam::${Account}:role/${RoleName}.
    • חשבון שירות: חשבון השירות שיצרתם בשלב יצירת חשבון שירות ב- Google Cloud.

  8. לוחצים על Create topic.

gcloud

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

    הפעלת Cloud Shell

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

  2. מריצים את הפקודה gcloud pubsub topics create:

    gcloud pubsub topics create TOPIC_ID \
      --aws-msk-ingestion-cluster-arn MSK_CLUSTER_ARN \
      --aws-msk-ingestion-topic MSK_TOPIC \
      --aws-msk-ingestion-aws-role-arn MSK_ROLE_ARN \
      --aws-msk-ingestion-service-account PUBSUB_SERVICE_ACCOUNT

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

    • ‫TOPIC_ID: השם או המזהה של נושא Pub/Sub.
    • ‫MSK_CLUSTER_ARN: ה-ARN של אשכול Amazon MSK שמתבצעת ממנו העברה ל-Pub/Sub. הפורמט של ה-ARN הוא: arn:aws:kafka:${Region}:${Account}:cluster/${ClusterName}/${ClusterId}.
    • ‫MSK_TOPIC: השם של נושא Amazon MSK Kafka שמתבצעת ממנו העברה אל Pub/Sub.
    • ‫MSK_ROLE_ARN: ה-ARN של תפקיד AWS. פורמט ה-ARN של התפקיד הוא: arn:aws:iam::${Account}:role/${RoleName}.
    • ‫PUBSUB_SERVICE_ACCOUNT: חשבון השירות שיצרתם בקטע יצירת חשבון שירות ב-Google Cloud.

C++‎

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של C++‎ במאמר התחלה מהירה: שימוש בספריות לקוח. מידע נוסף זמין במאמרי העזרה של Pub/Sub C++ API. 

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::TopicAdminClient client, std::string project_id,
   std::string topic_id, std::string const& cluster_arn,
   std::string const& msk_topic, std::string const& aws_role_arn,
   std::string const& gcp_service_account) {
  google::pubsub::v1::Topic request;
  request.set_name(
      pubsub::Topic(std::move(project_id), std::move(topic_id)).FullName());
  auto* aws_msk =
      request.mutable_ingestion_data_source_settings()->mutable_aws_msk();
  aws_msk->set_cluster_arn(cluster_arn);
  aws_msk->set_topic(msk_topic);
  aws_msk->set_aws_role_arn(aws_role_arn);
  aws_msk->set_gcp_service_account(gcp_service_account);

  auto topic = client.CreateTopic(request);
  // Note that kAlreadyExists is a possible error when the library retries.
  if (topic.status().code() == google::cloud::StatusCode::kAlreadyExists) {
    std::cout << "The topic already exists\n";
    return;
  }
  if (!topic) throw std::move(topic).status();

  std::cout << "The topic was successfully created: " << topic->DebugString()
            << "\n";
}

המשך

בדוגמה הבאה נעשה שימוש בגרסה הראשית של ספריית הלקוח Go Pub/Sub ‏ (v2). אם אתם עדיין משתמשים בספרייה של גרסה 1, עיינו במדריך להעברת נתונים לגרסה 2. כדי לראות רשימה של דוגמאות קוד מגרסה 1, אפשר לעיין ב דוגמאות הקוד שהוצאו משימוש.

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Go במאמר מדריך למתחילים: שימוש בספריות לקוח. מידע נוסף מופיע במסמכי העזר של ה-API של Pub/Sub Go.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
)

func createTopicWithAWSMSKIngestion(w io.Writer, projectID, topicID, clusterARN, mskTopic, awsRoleARN, gcpSA string) error {
	// projectID := "my-project-id"
	// topicID := "my-topic"

	// // AWS MSK ingestion settings.
	// clusterARN := "cluster-arn"
	// mskTopic := "msk-topic"
	// awsRoleARN := "aws-role-arn"
	// gcpSA := "gcp-service-account"

	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	topicpb := &pubsubpb.Topic{
		Name: fmt.Sprintf("projects/%s/topics/%s", projectID, topicID),
		IngestionDataSourceSettings: &pubsubpb.IngestionDataSourceSettings{
			Source: &pubsubpb.IngestionDataSourceSettings_AwsMsk_{
				AwsMsk: &pubsubpb.IngestionDataSourceSettings_AwsMsk{
					ClusterArn:        clusterARN,
					Topic:             mskTopic,
					AwsRoleArn:        awsRoleARN,
					GcpServiceAccount: gcpSA,
				},
			},
		},
	}
	topic, err := client.TopicAdminClient.CreateTopic(ctx, topicpb)
	if err != nil {
		return fmt.Errorf("CreateTopic: %w", err)
	}
	fmt.Fprintf(w, "Created topic with AWS MSK ingestion settings: %v\n", topic)
	return nil
}

Java

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Java במאמר תחילת העבודה המהירה: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub Java API. 


import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.pubsub.v1.IngestionDataSourceSettings;
import com.google.pubsub.v1.Topic;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;

public class CreateTopicWithAwsMskIngestionExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    // AWS MSK ingestion settings.
    String clusterArn = "cluster-arn";
    String mskTopic = "msk-topic";
    String awsRoleArn = "aws-role-arn";
    String gcpServiceAccount = "gcp-service-account";

    createTopicWithAwsMskIngestionExample(
        projectId, topicId, clusterArn, mskTopic, awsRoleArn, gcpServiceAccount);
  }

  public static void createTopicWithAwsMskIngestionExample(
      String projectId,
      String topicId,
      String clusterArn,
      String mskTopic,
      String awsRoleArn,
      String gcpServiceAccount)
      throws IOException {
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
      TopicName topicName = TopicName.of(projectId, topicId);

      IngestionDataSourceSettings.AwsMsk awsMsk =
          IngestionDataSourceSettings.AwsMsk.newBuilder()
              .setClusterArn(clusterArn)
              .setTopic(mskTopic)
              .setAwsRoleArn(awsRoleArn)
              .setGcpServiceAccount(gcpServiceAccount)
              .build();
      IngestionDataSourceSettings ingestionDataSourceSettings =
          IngestionDataSourceSettings.newBuilder().setAwsMsk(awsMsk).build();

      Topic topic =
          topicAdminClient.createTopic(
              Topic.newBuilder()
                  .setName(topicName.toString())
                  .setIngestionDataSourceSettings(ingestionDataSourceSettings)
                  .build());

      System.out.println("Created topic with AWS MSK ingestion settings: " + topic.getAllFields());
    }
  }
}

Node.js

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Node.js במאמר תחילת העבודה: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub Node.js API. 

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const clusterArn = 'arn:aws:kafka:...';
// const mskTopic = 'YOUR_MSK_TOPIC';
// const roleArn = 'arn:aws:iam:...';
// const gcpServiceAccount = 'ingestion-account@...';

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createTopicWithAwsMskIngestion(
  topicNameOrId,
  clusterArn,
  mskTopic,
  awsRoleArn,
  gcpServiceAccount,
) {
  // Creates a new topic with AWS MSK ingestion.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      awsMsk: {
        clusterArn,
        topic: mskTopic,
        awsRoleArn,
        gcpServiceAccount,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with AWS MSK ingestion.`);
}

Node.ts

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Node.js במאמר מדריך למתחילים: שימוש בספריות לקוח. מידע נוסף מופיע במאמרי העזרה של Pub/Sub Node.js API. 

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const clusterArn = 'arn:aws:kafka:...';
// const mskTopic = 'YOUR_MSK_TOPIC';
// const roleArn = 'arn:aws:iam:...';
// const gcpServiceAccount = 'ingestion-account@...';

// Imports the Google Cloud client library
import {PubSub} from '@google-cloud/pubsub';

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createTopicWithAwsMskIngestion(
  topicNameOrId: string,
  clusterArn: string,
  mskTopic: string,
  awsRoleArn: string,
  gcpServiceAccount: string,
) {
  // Creates a new topic with AWS MSK ingestion.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      awsMsk: {
        clusterArn,
        topic: mskTopic,
        awsRoleArn,
        gcpServiceAccount,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with AWS MSK ingestion.`);
}

Python

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

from google.cloud import pubsub_v1
from google.pubsub_v1.types import Topic
from google.pubsub_v1.types import IngestionDataSourceSettings

# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# cluster_arn = "your-cluster-arn"
# msk_topic = "your-msk-topic"
# aws_role_arn = "your-aws-role-arn"
# gcp_service_account = "your-gcp-service-account"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

request = Topic(
    name=topic_path,
    ingestion_data_source_settings=IngestionDataSourceSettings(
        aws_msk=IngestionDataSourceSettings.AwsMsk(
            cluster_arn=cluster_arn,
            topic=msk_topic,
            aws_role_arn=aws_role_arn,
            gcp_service_account=gcp_service_account,
        )
    ),
)

topic = publisher.create_topic(request=request)

print(f"Created topic: {topic.name} with AWS MSK Ingestion Settings")

מידע נוסף על ARNs זמין במאמרים Amazon Resource Names (ARNs)‎ ו-IAM Identifiers.

נתקלתם בבעיות? תוכלו להיעזר בקטע פתרון בעיות בנושא ייבוא מ-Amazon MSK.

עריכת נושאי הייבוא של Amazon MSK

כדי לערוך את הגדרות מקור הנתונים של נושא ייבוא ב-Amazon MSK, פועלים לפי השלבים הבאים:

המסוף

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

    לדף Topics

  2. לוחצים על נושא הייבוא של Amazon MSK.

  3. בדף הפרטים של הנושא, לוחצים על עריכה.

  4. מעדכנים את השדות שרוצים לשנות.

  5. לוחצים על עדכון.

gcloud

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

    הפעלת Cloud Shell

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

  2. מריצים את הפקודה gcloud pubsub topics update עם כל הדגלים שמוזכרים בדוגמה הבאה:

    gcloud pubsub topics update TOPIC_ID \
    --aws-msk-ingestion-cluster-arn MSK_CLUSTER_ARN \
    --aws-msk-ingestion-topic MSK_TOPIC \
    --aws-msk-ingestion-aws-role-arn MSK_ROLE_ARN \
    --aws-msk-ingestion-service-account PUBSUB_SERVICE_ACCOUNT

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

    • ‫TOPIC_ID: השם או המזהה של נושא ה-Pub/Sub.
    • ‫MSK_CLUSTER_ARN: ה-ARN של אשכול Amazon MSK שמתבצעת ממנו העברה ל-Pub/Sub. הפורמט של ה-ARN הוא: arn:aws:kafka:${Region}:${Account}:cluster/${ClusterName}/${ClusterId}.
    • ‫MSK_TOPIC: השם של נושא Amazon MSK Kafka שמתבצעת ממנו העברה ל-Pub/Sub.
    • ‫MSK_ROLE_ARN: ה-ARN של תפקיד AWS. הפורמט של ה-ARN של התפקיד הוא: arn:aws:iam::${Account}:role/${RoleName}.
    • ‫PUBSUB_SERVICE_ACCOUNT: חשבון השירות שיצרתם ביצירת חשבון שירות ב- Google Cloud.

מכסות ומגבלות

קצב העברת הנתונים של בעל התוכן הדיגיטלי לנושאי ייבוא מוגבל על ידי מכסת הפרסום של הנושא. מידע נוסף זמין במאמר מכסות ומגבלות ב-Pub/Sub.

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

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