Google menggunakan teknologi AI untuk menerjemahkan konten ke dalam bahasa pilihan Anda. Terjemahan AI mungkin mengandung kesalahan.

Membuat konektor Cloud Storage Sink

Konektor Cloud Storage Sink mengalirkan data dari Managed Service untuk Apache Kafka ke Cloud Storage, yang memungkinkan Anda menyimpan dan memproses data dalam volume besar secara hemat biaya dan skalabel.

Kasus penggunaan umum untuk konektor ini meliputi:

Penyerapan data lake. Menyimpan data Kafka di data lake untuk pengarsipan jangka panjang dan batch processing.
Mengarsipkan data untuk memenuhi persyaratan peraturan.

Sebelum memulai

Sebelum membuat konektor Cloud Storage Sink, pastikan Anda memiliki hal berikut:

Buat cluster Managed Service untuk Apache Kafka untuk cluster Connect Anda. Ini adalah cluster Kafka utama yang terkait dengan cluster Connect. Cluster ini juga merupakan cluster sumber yang membentuk salah satu ujung pipeline konektor.
Buat cluster Connect untuk menghosting konektor Cloud Storage Sink.
Buat bucket Cloud Storage untuk menyimpan data yang di-streaming dari Kafka.
Buat dan konfigurasikan topik Kafka dalam cluster sumber. Data berpindah dari topik Kafka ini ke bucket Cloud Storage tujuan.

Peran dan izin yang diperlukan

Untuk mendapatkan izin yang diperlukan guna membuat konektor, minta administrator untuk memberi Anda peran IAM Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) di project Anda. Untuk mengetahui informasi selengkapnya tentang cara memberikan peran, lihat Mengelola akses ke project, folder, dan organisasi.

Peran bawaan ini berisi izin yang diperlukan untuk membuat konektor. Untuk melihat izin yang benar-benar diperlukan, perluas bagian Izin yang diperlukan:

Izin yang diperlukan

Izin berikut diperlukan untuk membuat konektor:

Buat konektor: managedkafka.connectors.create

Anda mungkin juga bisa mendapatkan izin ini dengan peran khusus atau peran bawaan lainnya.

Memberikan izin untuk menulis ke bucket Cloud Storage

Agen Layanan Kafka Terkelola harus memiliki izin berikut untuk menulis pesan ke bucket Cloud Storage:

storage.objects.create
storage.objects.delete

Untuk memberikan izin ini, beri agen layanan peran Storage Object User (roles/storage.objectUser) pada project yang berisi bucket Cloud Storage.

Jika cluster Connect Anda berada di project yang berbeda dengan cluster Managed Service untuk Apache Kafka, lihat Membuat cluster Connect di project yang berbeda.

Membuat konektor Cloud Storage Sink

Untuk membuat konektor Cloud Storage Sink, lakukan langkah-langkah berikut.

Konsol

Di konsol Google Cloud , buka halaman Connect Clusters.

Buka Connect Clusters
Klik cluster Connect yang ingin Anda buatkan konektornya.

Halaman Connect cluster details akan ditampilkan.
Klik Create connector.

Halaman Create Kafka connector akan ditampilkan.
Untuk nama konektor, masukkan string.

Untuk panduan tentang cara memberi nama konektor, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka.
Untuk Connector plugin, pilih Cloud Storage Sink.
Tentukan Topik yang dapat Anda gunakan untuk melakukan streaming data.
Pilih Bucket Penyimpanan untuk menyimpan data.
(Opsional) Konfigurasikan setelan tambahan di bagian Configuration.
Pilih Task restart policy. Untuk mengetahui informasi selengkapnya, lihat Kebijakan memulai ulang tugas.
Klik Create.

gcloud

Jalankan perintah gcloud managed-kafka connectors create:

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

Ganti kode berikut:

CONNECTOR_ID: ID atau nama konektor.
CONNECT_CLUSTER_ID: ID cluster Connect tempat konektor dibuat.
LOCATION: Lokasi cluster Connect.
CONFIG_FILE: Jalur ke file konfigurasi YAML atau JSON lokal.

Berikut adalah contoh file konfigurasi untuk konektor 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"

Ganti kode berikut:

BUCKET_NAME: Nama bucket Cloud Storage yang akan ditulisi.
TOPIC_LIST: Daftar topik Kafka yang dipisahkan koma untuk dibaca. Daftar ini juga dapat menyertakan ekspresi reguler.

Terraform

Anda dapat menggunakan resource Terraform untuk membuat konektor.

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
}

Untuk mempelajari cara menerapkan atau menghapus konfigurasi Terraform, lihat Perintah dasar Terraform.

Go

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Go di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Managed Service for Apache Kafka Go API.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi(ADC). Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.

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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Java di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Managed Service for Apache Kafka Java API.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.


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

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Python di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Python API Managed Service untuk Apache Kafka.

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}")

Setelah membuat konektor, Anda dapat mengedit, menghapus, menjeda, menghentikan, atau memulai ulang konektor.

Mengonfigurasi konektor

Untuk mengetahui daftar lengkap properti yang khusus untuk konektor ini, lihat Konfigurasi dalam dokumentasi konektor Cloud Storage Sink.

Kolom output

Secara default, konektor hanya menulis nilai data dari setiap data Kafka. Untuk menyertakan kunci atau header data, tetapkan properti format.output.fields ke daftar kolom yang dipisahkan koma untuk disertakan. Kolom yang didukung adalah key, value, headers, dan timestamp. Contoh:

format.output.fields=key,value,headers

Format output

Konektor ini mendukung penulisan data Kafka dalam format CSV, JSON, dan Parquet. Bagian berikut menjelaskan cara mengonfigurasi format output.

Format CSV

Untuk menulis file CSV ke Cloud Storage, tetapkan format.output.type=csv. Gunakan konverter berikut:

Konverter kunci: org.apache.kafka.connect.storage.StringConverter atau org.apache.kafka.connect.json.JsonConverter
Konverter nilai: org.apache.kafka.connect.converters.ByteArrayConverter

Format JSON

Untuk menulis file berformat JSON ke Cloud Storage, tetapkan format.output.type ke salah satu nilai berikut:

json: Setiap file output berisi satu array objek JSON, dengan setiap objek JSON mewakili satu data Kafka.
jsonl: Konektor menulis setiap kumpulan data Kafka sebagai objek JSON pada baris terpisah.

Contoh berikut menunjukkan cara konektor menulis data Kafka dalam format JSON:

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

Jika Anda menetapkan pengonversi kunci atau nilai ke org.apache.kafka.connect.converters.ByteArrayConverter, konektor akan menulis kolom tersebut sebagai nilai berenkode base64.

Konektor tidak menyertakan skema JSON dalam output, meskipun value.converter.schemas.enable adalah true.

Format Parquet

Untuk menulis file Parquet ke Cloud Storage, tetapkan format.output.type=parquet.

Jika pengonversi kunci atau nilai adalah org.apache.kafka.connect.json.JsonConverter, data JSON harus menyertakan skema. Tetapkan properti schemas.enable konverter ke true.

Untuk data JSON, konektor menulis data sebagai jenis struct dalam file Parquet. Misalnya, jika nilai data Kafka memiliki kolom bernama userid, konektor akan menulis kolom bernama value.userid. Untuk meratakan nama kolom, tetapkan format.output.envelope=false. Jika format.output.envelope adalah false, format.output.fields harus menentukan satu kolom output.

Setelan file Cloud Storage

Setelan berikut mengontrol cara konektor menulis file ke Cloud Storage.

gcs.bucket.name: Wajib diisi. Menentukan nama bucket Cloud Storage tempat konektor menulis data.
file.compression.type: Menentukan jenis kompresi untuk file output. Contohnya adalah gzip, snappy, zstd, dan none. Nilai defaultnya adalah none.
file.name.prefix: Menentukan awalan yang akan ditambahkan ke nama setiap file. Nilai defaultnya kosong ("").
file.name.template: Menentukan template untuk nama file. Konektor ini mendukung variabel placeholder untuk template, seperti {{topic}} untuk nama topik dan {{partition}} untuk nomor partisi. Untuk mengetahui informasi selengkapnya, lihat dokumentasi konektor Cloud Storage Sink.
file.max.records: Jumlah maksimum data yang akan ditulis ke satu file. Jika nilainya 0, tidak ada nilai maksimum tertentu. Nilai defaultnya adalah 0.
file.max.bytes: Ukuran file maksimum yang akan ditulis, dalam byte. Jika nilainya 0, tidak ada nilai maksimum tertentu. Nilai defaultnya adalah 0.

Pengelompokan kumpulan data

Saat konektor Cloud Storage Sink menulis data Kafka, secara default konektor ini mengelompokkan data berdasarkan topik dan partisi. Misalnya, data dari partisi 0 topik my-topic ditulis ke file bernama my-topic-0-OFFSET, dengan OFFSET adalah offset awal.

Jika parameter file.name.template menyertakan variabel placeholder {{key}}, konektor akan mengelompokkan data berdasarkan kunci. Dalam mode ini, konektor menulis file terpisah untuk setiap kunci kumpulan data. Konektor menimpa setiap file dengan data terbaru untuk kunci tersebut.

Kredensial

Properti gcs.credentials.default harus ditetapkan ke true. Setelan ini memungkinkan konektor menemukan kredensial Google Cloud secara otomatis.

Apa langkah selanjutnya?

Apache Kafka® adalah merek dagang terdaftar dari Apache Software Foundation atau afiliasinya di Amerika Serikat dan/atau negara lainnya.

Membuat konektor Cloud Storage Sink Tetap teratur dengan koleksi Simpan dan kategorikan konten berdasarkan preferensi Anda.

Sebelum memulai

Peran dan izin yang diperlukan

Izin yang diperlukan

Memberikan izin untuk menulis ke bucket Cloud Storage

Membuat konektor Cloud Storage Sink

Konsol

gcloud

Terraform

Go

Java

Python

Mengonfigurasi konektor

Kolom output

Format output

Format CSV

Format JSON

Format Parquet

Setelan file Cloud Storage

Pengelompokan kumpulan data

Kredensial

Apa langkah selanjutnya?

Membuat konektor Cloud Storage Sink