카프카 커넥트

카프카 커넥트 (Kafka Connect)

카프카 커넥트(Kafka Connect)는 커넥터를 통해 Apache Kafka 안팎으로 데이터를 옮기는 인기 있는 프레임워크예요. 이 문서에서는 카프카 커넥트용 Apache Iceberg 싱크 커넥터에 대해 알려드릴게요. 카프카에서 아이스버그 테이블로 데이터를 쓰는 방법, 구성 속성, 카탈로그 설정, 다양한 라우팅 예시, 그리고 SMT(단순 메시지 변환)까지 폭넓게 살펴볼게요.

출처: 문서

본문

카프카 커넥트(Kafka Connect)는 커넥터를 통해 Apache Kafka 안팎으로 데이터를 옮기는 인기 있는 프레임워크예요. Kafka에서 S3로 데이터를 쓰는 S3 싱크나 관계형 데이터베이스의 변경 데이터 캡처(CDC) 레코드를 Kafka로 쓰는 Debezium 소스 커넥터 같은 다양한 커넥터를 사용할 수 있어요.

이것은 간단하고 분산된 아키텍처를 가져요. 클러스터는 여러 워커 프로세스로 구성되고, 커넥터는 이 프로세스에서 작업(task)을 실행해서 작업을 수행해요. 커넥터 배포는 구성 기반이므로 일반적으로 커넥터를 실행하는 데 코드를 작성할 필요가 없어요.

Apache Iceberg Sink Connector

Kafka Connect용 Apache Iceberg Sink Connector는 Kafka에서 아이스버그 테이블로 데이터를 쓰는 싱크 커넥터예요.

기능 (Features)

• 중앙 집중 아이스버그 커밋을 위한 커밋 조정(Commit coordination)
• 정확히 한 번(Exactly-once) 전달 의미론
• 다중 테이블 팬아웃(Multi-table fan-out)
• 자동 테이블 생성과 스키마 진화
• 아이스버그의 컬럼 매핑 기능을 통한 필드 이름 매핑

설치 (Installation)

커넥터 zip 아카이브는 아이스버그 빌드의 일부로 만들어져요. 다음으로 빌드를 실행할 수 있어요.

./gradlew -x test -x integrationTest clean build

zip 아카이브는 ./kafka-connect/kafka-connect-runtime/build/distributions 아래에서 찾을 수 있어요. Hive Metastore 클라이언트와 관련 의존성을 번들한 배포판과 번들하지 않은 배포판이 하나씩 있어요. 배포 아카이브를 모든 노드의 Kafka Connect 플러그인 디렉터리에 복사해요.

요구사항 (Requirements)

싱크는 KIP-447에 의존해서 정확히 한 번 의미론을 제공해요. 이는 Kafka 2.5 이상이 필요해요.

구성 (Configuration)

속성	설명
iceberg.tables	쉼표 구분 대상 테이블 목록
iceberg.tables.dynamic-enabled	routeField에 지정된 테이블로 라우팅하려면 true로 설정, routeRegex를 사용하지 않음. 기본값은 false
iceberg.tables.route-field	다중 테이블 팬아웃에서 레코드를 테이블로 라우팅하는 데 사용되는 필드 이름
iceberg.tables.default-commit-branch	커밋의 기본 브랜치, 지정하지 않으면 main 사용
iceberg.tables.default-id-columns	테이블에서 행을 식별하는 기본 쉼표 구분 컬럼 목록 (기본 키)
iceberg.tables.default-partition-by	테이블을 만들 때 사용할 기본 쉼표 구분 파티션 필드 이름 목록
iceberg.tables.auto-create-enabled	대상 테이블을 자동으로 만들려면 true로 설정, 기본값은 false
iceberg.tables.evolve-schema-enabled	누락된 레코드 필드를 테이블 스키마에 추가하려면 true로 설정, 기본값은 false
iceberg.tables.schema-force-optional	테이블 생성과 진화 중에 컬럼을 optional로 설정하려면 true로 설정, 스키마를 존중하려면 기본값은 false
iceberg.tables.schema-case-insensitive	테이블 컬럼을 대소문자 구분 없는 이름으로 찾으려면 true로 설정, 대소문자 구분의 기본값은 false
iceberg.tables.auto-create-props.*	auto-create 중 새 테이블에 설정되는 속성
iceberg.tables.write-props.*	아이스버그 작성자 초기화에 전달되는 속성, 이 속성이 우선함
iceberg.table..commit-branch	테이블별 커밋 브랜치, 지정하지 않으면 iceberg.tables.default-commit-branch 사용
iceberg.table..id-columns	테이블에서 행을 식별하는 쉼표 구분 컬럼 목록 (기본 키)
iceberg.table..partition-by	테이블을 만들 때 사용할 쉼표 구분 파티션 필드 목록
iceberg.table..route-regex	레코드의 routeField를 테이블과 일치시키는 데 사용되는 정규식
iceberg.control.topic	컨트롤 토픽 이름, 기본값은 control-iceberg
iceberg.control.group-id-prefix	컨트롤 컨슈머 그룹의 프리픽스, 기본값은 cg-control
iceberg.control.commit.interval-ms	커밋 간격(ms), 기본값은 300,000 (5 min)
iceberg.control.commit.timeout-ms	커밋 타임아웃 간격(ms), 기본값은 30,000 (30 sec)
iceberg.control.commit.threads	커밋에 사용할 스레드 수, 기본값은 (코어 수 * 2)
iceberg.coordinator.transactional.prefix	코디네이터 프로듀서에 사용할 트랜잭션 id의 프리픽스, 기본값은 프리픽스 없음/빈 값 사용
iceberg.catalog	카탈로그 이름, 기본값은 iceberg
iceberg.catalog.*	아이스버그 카탈로그 초기화에 전달되는 속성
iceberg.hadoop-conf-dir	지정하면 이 디렉터리의 Hadoop 구성 파일이 로드됨
iceberg.hadoop.*	Hadoop 구성에 전달되는 속성
iceberg.kafka.*	컨트롤 토픽 Kafka 클라이언트 초기화에 전달되는 속성

iceberg.tables.dynamic-enabled가 false(기본값)이면 iceberg.tables를 반드시 지정해야 해요. iceberg.tables.dynamic-enabled가 true이면 테이블 이름을 포함할 iceberg.tables.route-field를 반드시 지정해야 해요.

Kafka 구성 (Kafka configuration)

기본적으로 커넥터는 컨트롤 토픽에 연결하기 위해 워커 속성에서 Kafka 클라이언트 구성을 사용하려고 시도해요. 어떤 이유로든 그 구성을 읽을 수 없으면 iceberg.kafka.* 속성을 사용해 Kafka 클라이언트 설정을 명시적으로 설정할 수 있어요.

메시지 형식 (Message format)

메시지는 적절한 Kafka Connect 컨버터를 사용해 struct 또는 map으로 변환되어야 해요.

카탈로그 구성 (Catalog configuration)

iceberg.catalog.* 속성은 아이스버그 카탈로그에 연결하는 데 필요해요. 핵심 카탈로그 타입은 REST, Glue, DynamoDB, Hadoop, Nessie, JDBC, Hive, BigQuery Metastore를 포함해 기본 배포판에 포함돼 있어요. JDBC 드라이버는 기본 배포판에 포함되지 않으므로 필요하면 직접 포함해야 해요. Hive 카탈로그를 사용할 때는 Hive metastore 클라이언트를 포함한 배포판을 사용할 수 있고, 그렇지 않으면 직접 포함해야 해요.

카탈로그 타입을 설정하려면 iceberg.catalog.type을 rest, hive 또는 hadoop으로 설정할 수 있어요. 다른 카탈로그 타입의 경우 iceberg.catalog.catalog-impl을 카탈로그 클래스 이름으로 설정해야 해요.

REST 예시 (REST example)

"iceberg.catalog.type": "rest",
"iceberg.catalog.uri": "https://catalog-service",
"iceberg.catalog.credential": "<credential>",
"iceberg.catalog.warehouse": "<warehouse>",

Hive 예시 (Hive example)

참고: HMS 클라이언트를 포함한 배포판을 사용하거나(또는 HMS 클라이언트를 직접 포함) S3 스토리지에는 S3FileIO를, GCS에는 GCSFileIO를 사용해주세요 (HiveCatalog의 기본값은 HadoopFileIO예요).

"iceberg.catalog.type": "hive",
"iceberg.catalog.uri": "thrift://hive:9083",
"iceberg.catalog.io-impl": "org.apache.iceberg.aws.s3.S3FileIO",
"iceberg.catalog.warehouse": "s3a://bucket/warehouse",
"iceberg.catalog.client.region": "us-east-1",
"iceberg.catalog.s3.access-key-id": "<AWS access>",
"iceberg.catalog.s3.secret-access-key": "<AWS secret>",

Glue 예시 (Glue example)

"iceberg.catalog.catalog-impl": "org.apache.iceberg.aws.glue.GlueCatalog",
"iceberg.catalog.warehouse": "s3a://bucket/warehouse",
"iceberg.catalog.io-impl": "org.apache.iceberg.aws.s3.S3FileIO",

Nessie 예시 (Nessie example)

"iceberg.catalog.catalog-impl": "org.apache.iceberg.nessie.NessieCatalog",
"iceberg.catalog.uri": "http://localhost:19120/api/v2",
"iceberg.catalog.ref": "main",
"iceberg.catalog.warehouse": "s3a://bucket/warehouse",
"iceberg.catalog.io-impl": "org.apache.iceberg.aws.s3.S3FileIO",

BigQuery Metastore 예시 (BigQuery Metastore example)

"iceberg.catalog.catalog-impl": "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog",
"iceberg.catalog.gcp.bigquery.project-id": "my-project",
"iceberg.catalog.gcp.bigquery.location": "us-east1",
"iceberg.catalog.warehouse": "gs://bucket/warehouse",
"iceberg.catalog.io-impl": "org.apache.iceberg.gcp.gcs.GCSFileIO",
"iceberg.tables.auto-create-props.bq_connection": "projects/my-project/locations/us-east1/connections/my-connection",

참고 (Notes)

설정에 따라 iceberg.catalog.s3.endpoint, iceberg.catalog.s3.staging-dir, 또는 iceberg.catalog.s3.path-style-access도 설정해야 할 수 있어요. 카탈로그 구성에 대한 전체 내용은 아이스버그 문서를 참고해주세요.

Azure ADLS 구성 예시 (Azure ADLS configuration example)

ADLS를 사용할 때 Azure는 Java SDK에 AZURE_CLIENT_ID, AZURE_TENANT_ID, AZURE_CLIENT_SECRET을 전달해야 해요. Kafka Connect를 컨테이너에서 실행한다면 이 값들을 환경 변수로 주입해야 해요. 자세한 내용은 Java용 Azure Identity Client 라이브러리를 참고해주세요.

이 값들의 예시는 다음과 같아요.

AZURE_CLIENT_ID=e564f687-7b89-4b48-80b8-111111111111
AZURE_TENANT_ID=95f2f365-f5b7-44b1-88a1-111111111111
AZURE_CLIENT_SECRET="XXX"

여기서 CLIENT_ID는 App Registrations 아래 등록된 애플리케이션의 Application ID이고, TENANT_ID는 Azure 테넌트 속성에서 가져오고, CLIENT_SECRET은 App Registration을 선택한 후 "Manage" 아래 "Certificates & Secrets" 섹션에서 만들어져요. 하나를 생성하려면 중간 패널에서 "Client secrets"를 선택하고 "New client secret" 앞의 "+"를 선택해야 할 수 있어요. 이 변수를 Id가 아니라 Value로 설정해야 합니다.

또한 App Registration에 Storage Account의 Access Control(IAM)에서 역할 할당 "Storage Blob Data Contributor"가 부여되어야 하고, 그렇지 않으면 새 파일을 쓸 수 없어요.

그런 다음 커넥터 구성 안에 다음을 포함해야 해요.

"iceberg.catalog.type": "rest",
"iceberg.catalog.uri": "https://catalog:8181",
"iceberg.catalog.warehouse": "abfss://[email protected]/warehouse",
"iceberg.catalog.io-impl": "org.apache.iceberg.azure.adlsv2.ADLSFileIO",
"iceberg.catalog.include-credentials": "true"

여기서 storage-container-name은 Azure Storage Account 내의 컨테이너 이름이고, /warehouse는 (또는 iceberg.tables.auto-create-enabled=true이면) 기본적으로 Apache Iceberg 파일이 쓰여지는 컨테이너 내 위치이고, include-credentials 파라미터는 Azure Java 클라이언트 자격증명을 함께 전달해요. 이렇게 하면 Iceberg Sink 커넥터가 iceberg.catalog.uri의 REST 카탈로그 구현에 연결해서 ADLSv2 클라이언트에 필요한 Connection String을 얻어요.

Google GCS 구성 예시 (Google GCS configuration example)

기본적으로 Application Default Credentials(ADC)가 GCS에 연결하는 데 사용돼요. ADC가 어떻게 동작하는지에 대한 자세한 내용은 Google Cloud 문서에서 찾을 수 있어요.

"iceberg.catalog.type": "rest",
"iceberg.catalog.uri": "https://catalog:8181",
"iceberg.catalog.warehouse": "gs://bucket-name/warehouse",
"iceberg.catalog.io-impl": "org.apache.iceberg.gcp.gcs.GCSFileIO"

Hadoop 구성 (Hadoop configuration)

HDFS 또는 Hive를 사용할 때 싱크는 Hadoop 구성을 초기화해요. 먼저 클래스패스의 구성 파일이 로드돼요. 다음으로 iceberg.hadoop-conf-dir가 지정되면 그 위치에서 구성 파일이 로드돼요. 마지막으로 싱크 구성의 어떤 iceberg.hadoop.* 속성이든 적용돼요. 병합할 때 우선순위는 싱크 구성 > 구성 디렉터리 > 클래스패스예요.

예시 (Examples)

초기 설정 (Initial setup)

소스 토픽 (Source topic)

소스 토픽이 이미 존재하고 events라고 이름이 붙어 있다고 가정해요.

컨트롤 토픽 (Control topic)

Kafka 클러스터에 auto.create.topics.enable가 true(기본값)로 설정돼 있으면 컨트롤 토픽이 자동으로 만들어져요. 그렇지 않으면 토픽을 먼저 만들어야 해요. 기본 토픽 이름은 control-iceberg예요.

bin/kafka-topics.sh  \
  --command-config command-config.props \
  --bootstrap-server ${CONNECT_BOOTSTRAP_SERVERS} \
  --create \
  --topic control-iceberg \
  --partitions 1

참고: Confluent Cloud에서 실행되는 클러스터는 기본적으로 auto.create.topics.enable가 false로 설정돼 있어요.

아이스버그 카탈로그 구성 (Iceberg catalog configuration)

iceberg.catalog. 프리픽스가 있는 구성 속성은 아이스버그 카탈로그 초기화에 전달돼요. 특정 카탈로그를 구성하는 방법에 대한 자세한 내용은 아이스버그 문서를 참고해주세요.

단일 대상 테이블 (Single destination table)

이 예시는 모든 들어오는 레코드를 단일 테이블에 써요.

대상 테이블 만들기 (Create the destination table)

CREATE TABLE default.events (
    id STRING,
    type STRING,
    ts TIMESTAMP,
    payload STRING)
PARTITIONED BY (hours(ts))

커넥터 구성 (Connector config)

이 예시 구성은 아이스버그 REST 카탈로그에 연결해요.

{
  "name": "events-sink",
  "config": {
    "connector.class": "org.apache.iceberg.connect.IcebergSinkConnector",
    "tasks.max": "2",
    "topics": "events",
    "iceberg.tables": "default.events",
    "iceberg.catalog.type": "rest",
    "iceberg.catalog.uri": "https://localhost",
    "iceberg.catalog.credential": "<credential>",
    "iceberg.catalog.warehouse": "<warehouse name>"
  }
}

다중 테이블 팬아웃, 정적 라우팅 (Multi-table fan-out, static routing)

이 예시는 type이 list로 설정된 레코드를 default.events_list 테이블에 쓰고, type이 create로 설정된 레코드를 default.events_create 테이블에 써요. 다른 레코드는 건너뛰어져요.

두 개의 대상 테이블 만들기 (Create two destination tables)

CREATE TABLE default.events_list (
    id STRING,
    type STRING,
    ts TIMESTAMP,
    payload STRING)
PARTITIONED BY (hours(ts));

CREATE TABLE default.events_create (
    id STRING,
    type STRING,
    ts TIMESTAMP,
    payload STRING)
PARTITIONED BY (hours(ts));

커넥터 구성 (Connector config)

{
  "name": "events-sink",
  "config": {
    "connector.class": "org.apache.iceberg.connect.IcebergSinkConnector",
    "tasks.max": "2",
    "topics": "events",
    "iceberg.tables": "default.events_list,default.events_create",
    "iceberg.tables.route-field": "type",
    "iceberg.table.default.events_list.route-regex": "list",
    "iceberg.table.default.events_create.route-regex": "create",
    "iceberg.catalog.type": "rest",
    "iceberg.catalog.uri": "https://localhost",
    "iceberg.catalog.credential": "<credential>",
    "iceberg.catalog.warehouse": "<warehouse name>"
  }
}

다중 테이블 팬아웃, 동적 라우팅 (Multi-table fan-out, dynamic routing)

이 예시는 db_table 필드의 값에서 이름을 가져 온 테이블에 써요. 그 이름의 테이블이 없으면 레코드를 건너뛰어요. 예를 들어 레코드의 db_table 필드가 default.events_list로 설정되면 레코드는 default.events_list 테이블에 쓰여져요.

두 개의 대상 테이블 만들기 (Create two destination tables)

두 테이블 만들기는 위를 참고해주세요.

커넥터 구성 (Connector config)

{
  "name": "events-sink",
  "config": {
    "connector.class": "org.apache.iceberg.connect.IcebergSinkConnector",
    "tasks.max": "2",
    "topics": "events",
    "iceberg.tables.dynamic-enabled": "true",
    "iceberg.tables.route-field": "db_table",
    "iceberg.catalog.type": "rest",
    "iceberg.catalog.uri": "https://localhost",
    "iceberg.catalog.credential": "<credential>",
    "iceberg.catalog.warehouse": "<warehouse name>"
  }
}

Apache Iceberg Sink Connector용 SMT (SMTs for the Apache Iceberg Sink Connector)

이 프로젝트는 Iceberg 싱크 커넥터에서 사용하기 위해 Kafka 데이터를 변환할 때 유용할 수 있는 몇 가지 SMT를 포함해요.

CopyValue

(실험적)

CopyValue SMT는 한 필드의 값을 새 필드로 복사해요.

구성 (Configuration)

속성	설명
source.field	소스 필드 이름
target.field	대상 필드 이름

예시 (Example)

"transforms": "copyId",
"transforms.copyId.type": "org.apache.iceberg.connect.transforms.CopyValue",
"transforms.copyId.source.field": "id",
"transforms.copyId.target.field": "id_copy",

DmsTransform

(실험적)

DmsTransform SMT는 싱크의 CDC 기능에서 사용하기 위해 AWS DMS 형식의 메시지를 변환해요. data 요소 필드를 최상위로 승격하고 다음 메타데이터 필드를 추가해요: _cdc.op, _cdc.ts, _cdc.source.

구성 (Configuration)

SMT는 현재 구성이 없어요.

DebeziumTransform

(실험적)

DebeziumTransform SMT는 싱크의 CDC 기능에서 사용하기 위해 Debezium 형식의 메시지를 변환해요. before 또는 after 요소 필드를 최상위로 승격하고 다음 메타데이터 필드를 추가해요: _cdc.op, _cdc.ts, _cdc.offset, _cdc.source, _cdc.target, _cdc.key.

구성 (Configuration)

속성	설명
cdc.target.pattern	CDC target 필드 값 설정에 사용할 패턴, 기본값은 {db}.{table}

JsonToMapTransform

(실험적)

JsonToMapTransform SMT는 문자열을 Json 객체 페이로드로 파싱해서 스키마를 추론해요. 스키마리스 데이터(예: Kafka가 제공하는 JsonConverter가 만든 Map)를 위한 iceberg-kafka-connect 커넥터는 Map을 Iceberg Struct로 변환해요. JSON이 잘 구조화돼 있으면 이것은 괜찮지만, 동적으로 변하는 키를 가진 JSON 객체가 있으면 스키마 진화 때문에 Iceberg 테이블에서 컬럼 폭발이 일어날 수 있어요.

이 SMT는 JSON이 잘 구조화되지 않은 상황에서 데이터를 Iceberg로 가져와 쿼리 엔진이 더 관리하기 쉬운 형태로 추가 처리할 수 있게 하는 데 유용해요. 중첩 객체를 Map으로 변환하고 스키마에 Map 타입을 포함해요. 커넥터는 스키마를 존중해서 JSON 객체에 대해 Iceberg Map (String) 컬럼으로 아이스버그 테이블을 만들어요.

참고:

• 커넥터의 value.converter 설정으로 jsonConverter가 아닌 stringConverter를 반드시 사용해야 해요. 그 문자열에 JSON 객체({...})가 있다고 기대해요.
• 메시지 키, 톰스톤(tombstone), 헤더는 변환되지 않고 SMT에 의해 그대로 전달돼요.

구성 (Configuration)

속성	설명 (기본값)
json.root	(false) 루트에서 시작할지 여부를 나타내는 Boolean 값

transforms.IDENTIFIER_HERE.json.root는 가장 일관성이 없는 데이터를 위한 것이에요. 단일 필드 payload를 가진 Struct를 구성하고 Map<String, String> 스키마를 가져요.

transforms.IDENTIFIER_HERE.json.root가 false(기본값)이면 기본 타입과 배열 필드에 대해 추론된 스키마로 Struct를 구성해요. 중첩 객체는 Map<String, String> 타입의 필드가 돼요.

빈 배열과 빈 객체를 가진 키는 최종 스키마에서 걸러져요. 배열의 타입이 섞여 있지 않으면 배열은 타입이 지정되고, 섞여 있으면 문자열 배열로 변환돼요.

예시 json:

{
  "key": 1,
  "array": [1,"two",3],
  "empty_obj": {},
  "nested_obj": {"some_key": ["one", "two"]}
}

json.root가 true이면 다음이 됩니다.

SinkRecord.schema: 
  "payload" : (Optional) Map<String, String>

Sinkrecord.value (Struct): 
  "payload"  : Map(
    "key" : "1",
    "array" : "[1,"two",3]"
    "empty_obj": "{}"
    "nested_obj": "{"some_key":["one","two"]}"
  )

json.root가 false이면 다음이 됩니다.

SinkRecord.schema: 
  "key": (Optional) Int32,
  "array": (Optional) Array<String>,
  "nested_object": (Optional) Map<string, String>

SinkRecord.value (Struct):
  "key" 1, 
  "array" ["1", "two", "3"] 
  "nested_object" Map ("some_key" : "["one", "two"]") 

KafkaMetadataTransform

(실험적)

KafkaMetadata는 Kafka 메시지의 속성인 topic, partition, offset, timestamp를 주입해요.

구성 (Configuration)

속성	설명 (기본값)
field_name	(_kafka_metadata) 필드의 프리픽스
nested	(false) true이면 데이터를 struct에 중첩하고, 그렇지 않으면 프리픽스 필드로 최상위에 추가
external_field	(없음) 메타데이터에 상수 키, 값을 추가 (예: 클러스터 이름)

nested가 켜져 있으면:

_kafka_metadata.topic, _kafka_metadata.partition, _kafka_metadata.offset, _kafka_metadata.timestamp

nested가 꺼져 있으면:

_kafka_metadata_topic, _kafka_metadata_partition, _kafka_metadata_offset, _kafka_metadata_timestamp

MongoDebeziumTransform

(실험적)

MongoDebeziumTransform SMT는 before/after BSON 문자열이 있는 Mongo Debezium 형식의 메시지를 DebeziumTransform SMT가 기대하는 before/after 타입 Struct로 변환해요.

기본 카탈로그 타입이 mongodb 컬럼을 지원하지 않으면 컬럼 이름 변경은 (아직) 지원하지 않아요.

구성 (Configuration)

속성	설명
array_handling_mode	배열 처리 모드를 설정하는 array 또는 document

array 값(기본값)은 배열을 array 데이터 타입으로 인코딩해요. 주어진 배열 인스턴스의 모든 요소가 같은 타입인지 확인하는 것은 사용자의 책임이에요. 이 옵션은 제약적이지만 다운스트림 클라이언트가 배열을 쉽게 처리할 수 있게 해줘요.

document 값은 BSON 직렬화가 하는 것과 비슷한 방식으로 배열을 struct of structs로 변환해요. 메인 struct는 _0, _1, _2 등으로 명명된 필드를 포함하는데, 여기서 이름은 배열의 요소 인덱스를 나타내요. 각 요소는 주어진 필드의 값으로 전달돼요.

더 알아보기 (Learn more)

• 카탈로그 속성 (Catalog properties)
• 아이스버그 테이블 쓰기 (Iceberg writes)
• FileIO