‫Google משתמשת בטכנולוגיית AI כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

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

מחבר Cloud Storage Sink מעביר נתונים בסטרימינג מהשירות המנוהל ל-Apache Kafka אל 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

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

storage.objects.create
storage.objects.delete

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

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

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

כדי ליצור מחבר Cloud Storage Sink:

המסוף

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

מעבר אל Connect Clusters
לוחצים על אשכול Connect שרוצים ליצור לו את המחבר.

הדף Connect cluster details מוצג.
לוחצים על Create connector.

מוצג הדף Create Kafka connector.
מזינים מחרוזת בשם המחבר.

הנחיות למתן שמות למשאבים של שירות מנוהל ל-Apache Kafka
בשדה Connector plugin, בוחרים באפשרות Cloud Storage Sink.
מציינים את הנושאים שמהם אפשר להעביר נתונים בסטרימינג.
בוחרים את קטגוריית האחסון שבה רוצים לאחסן את הנתונים.
(אופציונלי) קובעים הגדרות נוספות בקטע Configuration.
בוחרים את מדיניות ההפעלה מחדש של המשימה. למידע נוסף, קראו את המאמר מדיניות ההפעלה מחדש של משימות.
לוחצים על יצירה.

gcloud

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

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

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

‫CONNECTOR_ID: המזהה או השם של המחבר.
‫CONNECT_CLUSTER_ID: המזהה של אשכול Connect שבו נוצר המחבר.
‫LOCATION: המיקום של אשכול Connect.
‫CONFIG_FILE: הנתיב לקובץ תצורה מקומי בפורמט YAML או JSON.

הנה דוגמה לקובץ תצורה של מחבר 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}")

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

הגדרת המחבר

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

שדות פלט

כברירת מחדל, המחבר כותב רק את ערך הרשומה מכל רשומת 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, שבו כל אובייקט מייצג רשומת 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 Sink.
‫file.max.records: המספר המקסימלי של רשומות שאפשר לכתוב בקובץ אחד. אם הערך הוא 0, אין מספר מקסימלי ספציפי. ערך ברירת המחדל הוא 0.
‫file.max.bytes: גודל הקובץ המקסימלי לכתיבה, בבייטים. אם הערך הוא 0, אין מספר מקסימלי ספציפי. ערך ברירת המחדל הוא 0.

קיבוץ רשומות

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

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

פרטי כניסה

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

מה השלב הבא?

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

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

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

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

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

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

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

המסוף

gcloud

Terraform

המשך

Java

Python

הגדרת המחבר

שדות פלט

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

פורמט CSV

פורמט JSON

פורמט Parquet

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

קיבוץ רשומות

פרטי כניסה

מה השלב הבא?

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