יצירת מחבר Cloud Storage Sink

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

תרחישים לדוגמה לשימוש במחברי יעד של Cloud Storage:

• הטמעה של אגם נתונים. אחסון נתוני Kafka באגם נתונים (data lake) לצורך ארכיון לטווח ארוך ועיבוד באצווה.

• העברת נתונים לארכיון כדי לעמוד בדרישות רגולטוריות.

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

לפני שיוצרים מחבר Cloud Storage Sink, צריך לוודא שיש לכם את הדברים הבאים:

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

• יוצרים אשכול Connect כדי לארח את מחבר Cloud Storage Sink.

• יוצרים קטגוריה של Cloud Storage לאחסון הנתונים שמוזרמים מ-Kafka.

• יוצרים ומגדירים נושא Kafka באשכול המקור. הנתונים עוברים מנושא Kafka הזה לקטגוריה של Cloud Storage.

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

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

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

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

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

• יצירת מחבר: managedkafka.connectors.create

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

מתן הרשאות כתיבה לקטגוריה של Cloud Storage

לחשבון השירות של Kafka המנוהל צריכות להיות ההרשאות הבאות כדי לכתוב הודעות לקטגוריה של Cloud Storage:

• storage.objects.create
• storage.objects.delete

מקצים לחשבון השירות את התפקיד Storage Object User (roles/storage.objectUser) בפרויקט שמכיל את הקטגוריה של Cloud Storage.

חשבון השירות המנוהל של Kafka הוא בפורמט הבא: [e-mail], כאשר PROJECT_NUMBER הוא מספר הפרויקט של אשכול Connect.

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

יצירת מחבר Cloud Storage Sink

כדי ליצור מחבר Cloud Storage Sink, מבצעים את השלבים הבאים.

המסוף

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

   מעבר אל Connect Clusters

2. לוחצים על Connect cluster (חיבור אשכול) שרוצים ליצור עבורו את המחבר.

   מוצג הדף Connect cluster details.

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

   מוצג הדף Create Kafka connector.

4. בשדה שם המחבר, מזינים מחרוזת.

   הנחיות לשמות של מחברים זמינות במאמר הנחיות לשמות של משאבים בשירות מנוהל ל-Apache Kafka.

5. בשדה Connector plugin, בוחרים באפשרות Cloud Storage Sink.

6. מציינים את הנושאים שמהם אפשר להזרים נתונים.

7. בוחרים את Storage Bucket לאחסון הנתונים.

8. (אופציונלי) מגדירים הגדרות נוספות בקטע הגדרות.

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

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

gcloud

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

   הפעלת Cloud Shell

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

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

   gcloud managed-kafka connectors create CONNECTOR_ID \
       --location=LOCATION \
       --connect-cluster=CONNECT_CLUSTER_ID \
       --config-file=CONFIG_FILE
   

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

   • ‫CONNECTOR_ID: המזהה או השם של המחבר. הנחיות לשמות של מחברים זמינות במאמר הנחיות לשמות של משאבים בשירות מנוהל ל-Apache Kafka. אי אפשר לשנות את השם של מחבר.

   • ‫LOCATION: המיקום שבו יוצרים את מחבר. המיקום הזה צריך להיות זהה למיקום שבו יצרתם את אשכול Connect.

   • ‫CONNECT_CLUSTER_ID: המזהה של אשכול Connect שבו נוצר המחבר.

   • ‫CONFIG_FILE: הנתיב לקובץ התצורה של המחבר בפורמט YAML.

   הנה דוגמה לקובץ תצורה של מחבר Cloud Storage Sink:

   connector.class: "io.aiven.kafka.connect.gcs.GcsSinkConnector"
   gcs.bucket.name: "BUCKET_NAME"
   gcs.credentials.default: "true"
   format.output.type: "json"
   key.converter: "org.apache.kafka.connect.storage.StringConverter"
   value.converter: "org.apache.kafka.connect.json.JsonConverter"
   value.converter.schemas.enable: "false"
   topics: "TOPIC_LIST"
   tasks.max: "3"
   

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

   • ‫BUCKET_NAME: השם של קטגוריית Cloud Storage שאליה ייכתבו הנתונים.

   • ‫TOPIC_LIST: רשימה מופרדת בפסיקים של נושאי Kafka לקריאה. הרשימה יכולה לכלול גם ביטויים רגולריים.

Terraform

אפשר להשתמש במשאב של Terraform כדי ליצור מחבר.

resource "google_managed_kafka_connector" "example-cloud-storage-sink-connector" {
  project         = data.google_project.default.project_id
  connector_id    = "my-gcs-sink-connector"
  connect_cluster = google_managed_kafka_connect_cluster.default.connect_cluster_id
  location        = "us-central1"

  configs = {
    "connector.class"                = "io.aiven.kafka.connect.gcs.GcsSinkConnector"
    "tasks.max"                      = "3"
    "topics"                         = "GMK_TOPIC_ID"
    "gcs.bucket.name"                = "GCS_BUCKET_NAME"
    "gcs.credentials.default"        = "true"
    "format.output.type"             = "json"
    "name"                           = "my-gcs-sink-connector"
    "value.converter"                = "org.apache.kafka.connect.json.JsonConverter"
    "value.converter.schemas.enable" = "false"
    "key.converter"                  = "org.apache.kafka.connect.storage.StringConverter"
  }
  provider = google-beta
}

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

המשך

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

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

import (
	"context"
	"fmt"
	"io"

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

// createCloudStorageSinkConnector creates a Cloud Storage Sink connector.
func createCloudStorageSinkConnector(w io.Writer, projectID, region, connectClusterID, connectorID, topics, gcsBucketName, tasksMax, formatOutputType, valueConverter, valueConverterSchemasEnable, keyConverter, gcsCredentialsDefault string, opts ...option.ClientOption) error {
	// TODO(developer): Update with your config values. Here is a sample configuration:
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "GCS_SINK_CONNECTOR_ID"
	// topics := "GMK_TOPIC_ID"
	// gcsBucketName := "GCS_BUCKET_NAME"
	// tasksMax := "3"
	// formatOutputType := "json"
	// valueConverter := "org.apache.kafka.connect.json.JsonConverter"
	// valueConverterSchemasEnable := "false"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	// gcsCredentialsDefault := "true"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	parent := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s", projectID, region, connectClusterID)

	config := map[string]string{
		"connector.class":                "io.aiven.kafka.connect.gcs.GcsSinkConnector",
		"tasks.max":                      tasksMax,
		"topics":                         topics,
		"gcs.bucket.name":                gcsBucketName,
		"gcs.credentials.default":        gcsCredentialsDefault,
		"format.output.type":             formatOutputType,
		"name":                           connectorID,
		"value.converter":                valueConverter,
		"value.converter.schemas.enable": valueConverterSchemasEnable,
		"key.converter":                  keyConverter,
	}

	connector := &managedkafkapb.Connector{
		Name:    fmt.Sprintf("%s/connectors/%s", parent, connectorID),
		Configs: config,
	}

	req := &managedkafkapb.CreateConnectorRequest{
		Parent:      parent,
		ConnectorId: connectorID,
		Connector:   connector,
	}

	resp, err := client.CreateConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Created Cloud Storage sink connector: %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.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ConnectClusterName;
import com.google.cloud.managedkafka.v1.Connector;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.CreateConnectorRequest;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class CreateCloudStorageSinkConnector {

  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 connectClusterId = "my-connect-cluster";
    String connectorId = "my-gcs-sink-connector";
    String bucketName = "my-gcs-bucket";
    String kafkaTopicName = "kafka-topic";
    String connectorClass = "io.aiven.kafka.connect.gcs.GcsSinkConnector";
    String maxTasks = "3";
    String gcsCredentialsDefault = "true";
    String formatOutputType = "json";
    String valueConverter = "org.apache.kafka.connect.json.JsonConverter";
    String valueSchemasEnable = "false";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    createCloudStorageSinkConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        bucketName,
        kafkaTopicName,
        connectorClass,
        maxTasks,
        gcsCredentialsDefault,
        formatOutputType,
        valueConverter,
        valueSchemasEnable,
        keyConverter);
  }

  public static void createCloudStorageSinkConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String bucketName,
      String kafkaTopicName,
      String connectorClass,
      String maxTasks,
      String gcsCredentialsDefault,
      String formatOutputType,
      String valueConverter,
      String valueSchemasEnable,
      String keyConverter)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("connector.class", connectorClass);
    configMap.put("tasks.max", maxTasks);
    configMap.put("topics", kafkaTopicName);
    configMap.put("gcs.bucket.name", bucketName);
    configMap.put("gcs.credentials.default", gcsCredentialsDefault);
    configMap.put("format.output.type", formatOutputType);
    configMap.put("name", connectorId);
    configMap.put("value.converter", valueConverter);
    configMap.put("value.converter.schemas.enable", valueSchemasEnable);
    configMap.put("key.converter", keyConverter);

    Connector connector = Connector.newBuilder()
        .setName(
            ConnectorName.of(projectId, region, connectClusterId, connectorId).toString())
        .putAllConfigs(configMap)
        .build();

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
      CreateConnectorRequest request = CreateConnectorRequest.newBuilder()
          .setParent(ConnectClusterName.of(projectId, region, connectClusterId).toString())
          .setConnectorId(connectorId)
          .setConnector(connector)
          .build();

      // This operation is being handled synchronously.
      Connector response = managedKafkaConnectClient.createConnector(request);
      System.out.printf("Created Cloud Storage Sink connector: %s\n", response.getName());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.createConnector got err: %s\n", e.getMessage());
    }
  }
}

Python

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

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

from google.api_core.exceptions import GoogleAPICallError
from google.cloud.managedkafka_v1.services.managed_kafka_connect import (
    ManagedKafkaConnectClient,
)
from google.cloud.managedkafka_v1.types import Connector, CreateConnectorRequest

connect_client = ManagedKafkaConnectClient()
parent = connect_client.connect_cluster_path(project_id, region, connect_cluster_id)

configs = {
    "connector.class": "io.aiven.kafka.connect.gcs.GcsSinkConnector",
    "tasks.max": tasks_max,
    "topics": topics,
    "gcs.bucket.name": gcs_bucket_name,
    "gcs.credentials.default": "true",
    "format.output.type": format_output_type,
    "name": connector_id,
    "value.converter": value_converter,
    "value.converter.schemas.enable": value_converter_schemas_enable,
    "key.converter": key_converter,
}

connector = Connector()
connector.name = connector_id
connector.configs = configs

request = CreateConnectorRequest(
    parent=parent,
    connector_id=connector_id,
    connector=connector,
)

try:
    operation = connect_client.create_connector(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    response = operation.result()
    print("Created Connector:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e}")

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

הגדרת המחבר

רשימה מלאה של המאפיינים הספציפיים למחבר הזה מופיעה במאמר הגדרה במסמכי התיעוד של מחבר Cloud Storage Sink.

שדות פלט

כברירת מחדל, המחבר כותב רק את ערך הרשומה מכל רשומה של Kafka. כדי לכלול מפתחות או כותרות של רשומות, מגדירים את המאפיין format.output.fields לרשימה מופרדת בפסיקים של השדות שרוצים לכלול. השדות הנתמכים הם key,‏ value,‏ headers ו-timestamp. לדוגמה:

format.output.fields=key,value,headers

פורמטים של פלט

המחבר תומך בכתיבת רשומות Kafka בפורמטים CSV,‏ JSON ו-Parquet. בקטעים הבאים מוסבר איך להגדיר את פורמט הפלט.

פורמט CSV

כדי לכתוב קובצי CSV ל-Cloud Storage, מגדירים את format.output.type=csv. אפשר להשתמש בממירים הבאים:

• מקש המרה: org.apache.kafka.connect.storage.StringConverter או org.apache.kafka.connect.json.JsonConverter

• ממיר ערכים: org.apache.kafka.connect.converters.ByteArrayConverter

פורמט JSON

כדי לכתוב קבצים בפורמט JSON ב-Cloud Storage, מגדירים את format.output.type לאחד מהערכים הבאים:

• ‫json: כל קובץ פלט מכיל מערך יחיד של אובייקטים בפורמט JSON, כאשר כל אובייקט בפורמט JSON מייצג רשומה אחת של Kafka.

• ‫jsonl: המחבר כותב כל רשומה של Kafka כאובייקט JSON בשורה נפרדת.

בדוגמה הבאה אפשר לראות איך המחבר כותב רשומת Kafka בפורמט JSON:

{"headers":[{"key":"k1","value":"v1"},{"key":"k2","value":"v2"}],"value":"Hello world","key":"my-key"}

אם מגדירים את ממיר המפתח או הערך ל-org.apache.kafka.connect.converters.ByteArrayConverter, המחבר כותב את השדה הזה כערך מקודד בפורמט Base64.

המחבר לא כולל סכימת JSON בפלט, גם אם value.converter.schemas.enable הוא true.

פורמט Parquet

כדי לכתוב קובצי Parquet ל-Cloud Storage, מגדירים את הערך של format.output.type=parquet.

אם ממיר המפתח או הערך הוא org.apache.kafka.connect.json.JsonConverter, נתוני ה-JSON צריכים לכלול סכימה. מגדירים את המאפיין schemas.enable של הכלי להמרה לערך true.

לגבי נתוני JSON, המחבר כותב את הנתונים כסוג struct בקובץ Parquet. לדוגמה, אם לערך הרשומה ב-Kafka יש שדה בשם userid, המחבר כותב עמודה בשם value.userid. כדי להסיר את ההיררכיה משמות העמודות, מגדירים את format.output.envelope=false. אם הערך של format.output.envelope הוא false, צריך לציין ב-format.output.fields שדה פלט יחיד.

הגדרות הקבצים ב-Cloud Storage

ההגדרות הבאות קובעות איך המחבר כותב קבצים ב-Cloud Storage.

• ‫gcs.bucket.name: שדה חובה. מציינת את השם של הקטגוריה ב-Cloud Storage שבה המחבר כותב נתונים.

• ‫file.compression.type: מציין את סוג הדחיסה של קובצי הפלט. דוגמאות: gzip, snappy, zstd ו-none. ערך ברירת המחדל הוא none.

• ‫file.name.prefix: מציין קידומת שתוסף לשם של כל קובץ. ערך ברירת המחדל הוא ריק ("").

• ‫file.name.template: מציינת תבנית לשם הקובץ. המחבר תומך במשתני placeholder לתבנית, כמו {{topic}} לשם הנושא ו-{{partition}} למספר המחיצה. מידע נוסף זמין במאמרי העזרה בנושא מחבר יעד של Cloud Storage.

קיבוץ רשומות

כשמחבר Cloud Storage Sink כותב רשומות של Kafka, כברירת מחדל הוא מקבץ את הרשומות לפי נושא ומחיצה. לדוגמה, רשומות ממחיצה 0 בנושא my-topic נכתבות לקבצים בשם my-topic-0-OFFSET, כאשר OFFSET הוא ההיסטור התחילי.

אם הפרמטר file.name.template כולל את משתנה הפלייסולדר {{key}}, הרשומות של קבוצות המחברים מסודרות לפי מפתח. במצב הזה, המחבר כותב קובץ נפרד לכל מפתח רשומה. המחבר מחליף כל קובץ ברשומה האחרונה של המפתח הזה.

פרטי כניסה

הערך של המאפיין gcs.credentials.default חייב להיות true. ההגדרה הזו מאפשרת למחבר לגלות Google Cloud פרטי כניסה באופן אוטומטי.

מה השלב הבא?

• פתרון בעיות בחיבור Cloud Storage Sink

• השהיה, הפעלה מחדש, עצירה והפעלה מחדש של מחבר

• רשימת כל המחברים

• הצגת הפרטים של מחבר

• עדכון מחבר

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