Google은 AI 기술을 사용하여 콘텐츠를 사용자의 기본 언어로 번역합니다. AI 번역에는 오류가 있을 수 있습니다.

Cloud Storage 싱크 커넥터 만들기

Cloud Storage 싱크 커넥터는 Kafka에서 Cloud Storage 버킷으로 데이터를 스트리밍하므로 비용 효율적이고 확장 가능한 방식으로 대용량 데이터를 저장하고 처리할 수 있습니다.

Cloud Storage 싱크 커넥터의 사용 사례는 다음과 같습니다.

데이터 레이크 수집. 장기 보관 및 일괄 처리를 위해 데이터 레이크에 Kafka 데이터를 저장합니다.
규정 요구사항을 충족하기 위해 데이터를 보관합니다.

시작하기 전에

Cloud Storage 싱크 커넥터를 만들기 전에 다음이 있는지 확인하세요.

Connect 클러스터의 Managed Service for Apache Kafka 클러스터를 만듭니다. 이는 Connect 클러스터와 연결된 기본 Kafka 클러스터입니다. 또한 커넥터 파이프라인의 한쪽 끝을 형성하는 소스 클러스터이기도 합니다.
Connect 클러스터 를 만들어 Cloud Storage 싱크 커넥터를 호스팅합니다.
Kafka에서 스트리밍된 데이터를 저장할 Cloud Storage 버킷을 만듭니다.
Kafka 주제를 만들고 구성합니다 소스 클러스터 내에서. 데이터는 이 Kafka 주제에서 대상 Cloud Storage 버킷으로 이동합니다.

필수 역할 및 권한

커넥터를 만드는 데 필요한 권한을 얻으려면 관리자에게 프로젝트에 대한 관리형 Kafka 커넥터 편집자 (roles/managedkafka.connectorEditor) IAM 역할을 부여해 달라고 요청하세요. 역할 부여에 대한 자세한 내용은 프로젝트, 폴더, 조직에 대한 액세스 관리를 참조하세요.

이 사전 정의된 역할에는 커넥터를 만드는 데 필요한 권한이 포함되어 있습니다. 필요한 정확한 권한을 보려면 필수 권한 섹션을 펼치세요.

필수 권한

커넥터를 만들려면 다음 권한이 필요합니다.

커넥터 만들기: managedkafka.connectors.create

커스텀 역할이나 다른 사전 정의된 역할을 사용하여 이 권한 을 부여받을 수도 있습니다.

Cloud Storage 버킷에 쓸 수 있는 권한 부여

관리형 Kafka 서비스 계정에는 Cloud Storage 버킷에 메시지를 쓰기 위한 다음과 같은 권한이 있어야 합니다.

storage.objects.create
storage.objects.delete

Cloud Storage 버킷이 포함된 프로젝트에서 서비스 계정에 스토리지 객체 사용자 (roles/storage.objectUser) 역할을 부여합니다.

관리형 Kafka 서비스 계정의 형식은 service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, 입니다. 여기서 PROJECT_NUMBER는 Connect 클러스터의 프로젝트 번호입니다.

Connect 클러스터가 Managed Service for Apache Kafka 클러스터와 다른 프로젝트에 있는 경우 다른 프로젝트에서 Connect 클러스터 만들기를 참조하세요.

Cloud Storage 싱크 커넥터 만들기

Cloud Storage 싱크 커넥터를 만들려면 다음 단계를 수행하세요.

콘솔

콘솔 Google Cloud 에서 Connect 클러스터 페이지로 이동합니다.

Connect 클러스터로 이동
커넥터를 만들려는 Connect 클러스터를 클릭합니다.

Connect 클러스터 세부정보 페이지가 표시됩니다.
커넥터 만들기 를 클릭합니다.

Kafka 커넥터 만들기 페이지가 표시됩니다.
커넥터 이름에 문자열을 입력합니다.

커넥터 이름을 지정하는 방법에 대한 가이드라인은 Managed Service for Apache Kafka 리소스 이름 지정 가이드라인을 참조하세요.
커넥터 플러그인에서 Cloud Storage 싱크를 선택합니다.
데이터를 스트리밍할 수 있는 주제 를 지정합니다.
데이터를 저장할 스토리지 버킷 을 선택합니다.
(선택사항) 구성 섹션에서 추가 설정을 구성합니다.
작업 재시작 정책 을 선택합니다. 자세한 내용은 작업 재시작 정책을 참조하세요.
만들기 를 클릭합니다.

gcloud

콘솔 Google Cloud 에서 Cloud Shell을 활성화합니다.

Cloud Shell 활성화

Google Cloud 콘솔 하단에 Cloud Shell 세션이 시작되고 명령줄 프롬프트가 표시됩니다. Cloud Shell은 Google Cloud CLI가 사전 설치된 셸 환경으로, 현재 프로젝트의 값이 이미 설정되어 있습니다. 세션이 초기화되는 데 몇 초 정도 걸릴 수 있습니다.
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: 커넥터의 ID 또는 이름입니다. 커넥터 이름을 지정하는 방법에 대한 가이드라인은 Managed Service for Apache Kafka 리소스 이름 지정 가이드라인을 참조하세요. 커넥터 이름은 변경할 수 없습니다.
- LOCATION: 커넥터를 만드는 위치입니다. Connect 클러스터를 만든 위치와 동일해야 합니다.
- CONNECT_CLUSTER_ID: 커넥터가 생성되는 Connect 클러스터의 ID입니다.
- CONFIG_FILE: 커넥터의 YAML 구성 파일 경로입니다.
다음은 Cloud Storage 싱크 커넥터의 구성 파일 예시입니다.
```
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

이 샘플을 시도하기 전에 클라이언트 라이브러리 설치의 Go 설정 안내를 따르세요. 자세한 내용은 Managed Service for Apache Kafka Go API 참고 문서를 참조하세요.

Managed Service for Apache Kafka에 인증하려면 애플리케이션 기본 사용자 인증 정보(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 설정 안내를 따르세요. 자세한 내용은 Managed Service for Apache Kafka Java API 참고 문서를 참조하세요.

Managed Service for Apache Kafka에 인증하려면 애플리케이션 기본 사용자 인증 정보를 설정합니다. 자세한 내용은 로컬 개발 환경의 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 설정 안내를 따르세요. 자세한 내용은 Managed Service for Apache Kafka Python API 참고 문서를 참조하세요.

Managed Service for Apache Kafka에 인증하려면 애플리케이션 기본 사용자 인증 정보를 설정합니다. 자세한 내용은 로컬 개발 환경의 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 싱크 커넥터 문서의 구성을 참조하세요.

출력란

기본적으로 커넥터는 각 Kafka 레코드의 레코드 값만 씁니다. 레코드 키 또는 헤더를 포함하려면 format.output.fields 속성을 포함할 필드의 쉼표로 구분된 목록으로 설정합니다. 지원되는 필드는 key, value, headers, timestamp입니다. 예를 들면 다음과 같습니다.

format.output.fields=key,value,headers

출력 형식

커넥터는 CSV, JSON, Parquet 형식으로 Kafka 레코드를 쓰는 것을 지원합니다. 다음 섹션에서는 출력 형식을 구성하는 방법을 설명합니다.

CSV 형식

Cloud Storage에 CSV 파일을 쓰려면 format.output.type=csv를 설정합니다. 다음 변환기를 사용합니다.

키 변환기: org.apache.kafka.connect.storage.StringConverter 또는 org.apache.kafka.connect.json.JsonConverter
값 변환기: org.apache.kafka.connect.converters.ByteArrayConverter

JSON 형식

Cloud Storage에 JSON 형식 파일을 쓰려면 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로 인코딩된 값으로 씁니다.

value.converter.schemas.enable이 true인 경우에도 커넥터는 출력에 JSON 스키마를 포함하지 않습니다.

Parquet 형식

Cloud Storage에 Parquet 파일을 쓰려면 format.output.type=parquet를 설정합니다.

키 또는 값 변환기가 org.apache.kafka.connect.json.JsonConverter인 경우 JSON 데이터에 스키마가 포함되어야 합니다. 변환기의 schemas.enable 속성을 true로 설정합니다.

JSON 데이터의 경우 커넥터는 데이터를 Parquet 파일의 struct 유형으로 씁니다. 예를 들어 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: 파일 이름의 템플릿을 지정합니다. 커넥터 는 주제 이름의 {{topic}} 및 파티션 번호의 {{partition}}와 같은 템플릿의 자리표시자 변수를 지원합니다. 자세한 내용은 다음 Cloud Storage 싱크 커넥터 문서를 참조하세요.
file.max.records: 단일 파일에 쓸 최대 레코드 수입니다. 값이 0이면 특정 최댓값이 없습니다. 기본값은 0입니다.
file.max.bytes: 쓸 최대 파일 크기(바이트)입니다. 값이 0이면 특정 최댓값이 없습니다. 기본값은 0입니다.

레코드 그룹화

Cloud Storage 싱크 커넥터가 Kafka 레코드를 쓸 때 기본적으로 레코드를 주제 및 파티션별로 그룹화합니다. 예를 들어 주제 my-topic의 파티션 0의 레코드는 my-topic-0-OFFSET이라는 파일에 기록됩니다. 여기서 OFFSET은 시작 오프셋입니다.

file.name.template 매개변수에 {{key}} 자리표시자 변수가 포함되어 있으면 커넥터는 레코드를 키별로 그룹화합니다. 이 모드에서 커넥터는 각 레코드 키에 대해 별도의 파일을 씁니다. 커넥터는 각 파일을 해당 키의 최신 레코드로 덮어씁니다.

사용자 인증 정보

gcs.credentials.default 속성은 true로 설정해야 합니다. 이 설정을 사용하면 커넥터가 사용자 인증 정보를 Google Cloud 자동으로 검색할 수 있습니다.

다음 단계

Apache Kafka®는 미국 및/또는 다른 국가에서 사용되는 Apache Software Foundation 또는 해당 계열사의 등록 상표입니다.

Cloud Storage 싱크 커넥터 만들기 컬렉션을 사용해 정리하기 내 환경설정을 기준으로 콘텐츠를 저장하고 분류하세요.

시작하기 전에

필수 역할 및 권한

필수 권한

Cloud Storage 버킷에 쓸 수 있는 권한 부여

Cloud Storage 싱크 커넥터 만들기

콘솔

gcloud

Terraform

Go

자바

Python

커넥터 구성

출력란

출력 형식

CSV 형식

JSON 형식

Parquet 형식

Cloud Storage 파일 설정

레코드 그룹화

사용자 인증 정보

다음 단계

Cloud Storage 싱크 커넥터 만들기