플링크 구성

플링크 구성 (Flink Configuration)

이 문서에서는 Flink에서 아이스버그 카탈로그를 구성하는 방법과 런타임 읽기·쓰기 옵션을 알려드릴게요. CREATE CATALOG로 카탈로그를 만들고 설정하는 방법부터, Flink SQL 힌트나 Flink 구성, 테이블 속성으로 전달하는 각종 읽기·쓰기 옵션의 기본값과 설명까지 표로 정리해드릴게요.

출처: 문서

본문

카탈로그 구성 (Catalog Configuration)

다음 쿼리를 실행해서 카탈로그를 만들고 이름을 붙여요(<catalog_name>은 카탈로그 이름으로, <config_key>=<config_value>는 카탈로그 구현 구성으로 바꿔주세요).

CREATE CATALOG <catalog_name> WITH (
  'type'='iceberg',
  `<config_key>`=`<config_value>`
); 

다음 속성들은 전역적으로 설정할 수 있고 특정 카탈로그 구현에 국한되지 않아요.

Hive 카탈로그를 사용할 때 다음 속성들을 설정할 수 있어요.

Hadoop 카탈로그를 사용할 때 다음 속성들을 설정할 수 있어요.

REST 카탈로그를 사용할 때 다음 속성들을 설정할 수 있어요.

런타임 구성 (Runtime configuration)

읽기 옵션 (Read options)

Flink 읽기 옵션은 Flink IcebergSource를 구성할 때 전달돼요.

IcebergSource.forRowData()
    .tableLoader(TableLoader.fromCatalog(...))
    .assignerFactory(new SimpleSplitAssignerFactory())
    .streaming(true)
    .streamingStartingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_SNAPSHOT_ID)
    .startSnapshotId(3821550127947089987L)
    .monitorInterval(Duration.ofMillis(10L)) // or .set("monitor-interval", "10s") \ set(FlinkReadOptions.MONITOR_INTERVAL, "10s")
    .build()

Flink SQL의 경우 읽기 옵션은 SQL 힌트로 전달할 수 있어요.

SELECT * FROM tableName /*+ OPTIONS('monitor-interval'='10s') */
...

옵션은 Flink 구성을 통해서도 전달할 수 있고, 이 경우 현재 세션에 적용돼요. 모든 옵션이 이 모드를 지원하는 것은 아니라는 점에 유의해주세요.

env.getConfig()
    .getConfiguration()
    .set(FlinkReadOptions.SPLIT_FILE_OPEN_COST_OPTION, 1000L);
...

읽기 옵션이 가장 높은 우선순위를 갖고, 그 다음 Flink 구성, 그다음 테이블 속성이에요.

쓰기 옵션 (Write options)

Flink 쓰기 옵션은 FlinkSink를 구성할 때 전달돼요.

FlinkSink.Builder builder = FlinkSink.forRow(dataStream, SimpleDataUtil.FLINK_SCHEMA)
    .table(table)
    .tableLoader(tableLoader)
    .set("write-format", "orc")
    .set(FlinkWriteOptions.OVERWRITE_MODE, "true");

Flink SQL의 경우 쓰기 옵션은 SQL 힌트로 전달할 수 있어요.

INSERT INTO tableName /*+ OPTIONS('upsert-enabled'='true') */
...

범위 분포 통계 유형 (Range distribution statistics type)

구성 값은 Map, Sketch, Auto라는 enum 타입이에요.

• Map: 모든 개별 키에 대한 정확한 샘플링 개수를 수집해요. 낮은 카디널리티(수백~수천) 시나리오에 사용해야 해요. Sketch: reservoir sampling을 통해 균일한 무작위 샘플링을 구성해요. 메모리 사용량이 낮게 유지되므로 높은 카디널리티(수백만) 시나리오에 잘 맞아요. Auto: Map 통계로 시작해요. 하지만 카디널리티가 임계값(현재 10,000)보다 높게 감지되면 통계가 자동으로 Sketch로 전환돼요.

범위 분포 정렬 키 기본 가중치 (Range distribution sort key base weight)

range-distribution-sort-key-base-weight: 0.0.

정렬 순서에 파티션 컬럼이 포함되면 각 정렬 키는 하나의 파티션과 데이터 파일에 매핑돼요. 이 상대 가중치는 트래픽이 낮은 정렬 키에 너무 많은 작은 파일을 배치하는 것을 피할 수 있어요. 이것은 각 정렬 키의 최소 가중치를 정의하는 double 값이에요. 0.02는 각 키가 작성자 작업별 목표 트래픽 가중치의 2%를 기본 가중치로 가진다는 뜻이에요.

예를 들어 싱크 아이스버그 테이블이 이벤트 시간으로 매일 파티셔닝된다고 해볼게요. 데이터 스트림에 지금부터 180일 전까지의 이벤트가 포함된다고 가정해요. 이벤트 시간으로 볼 때 서로 다른 날짜들에 걸친 트래픽 가중치 분포는 전형적으로 긴 꼬리(long tail) 패턴을 가져요. 현재 날짜가 가장 많은 트래픽을 포함하고, 오래된 날짜들(긴 꼬리)은 점점 더 적은 트래픽을 포함해요. 작성자 병렬도가 10이라고 가정해요. 180일 전체의 총 가중치는 10,000이에요. 작성자 작업별 목표 트래픽 가중치는 1,000이 돼요. 가장 오래된 150일의 가중치 합이 1,000이라고 가정해요. 일반적으로 범위 파티셔너는 가장 오래된 150일을 하나의 작성자 작업에 넣어요. 그 작성자 작업은 150개의 작은 파일(하루에 하나)을 쓰게 돼요. 150개의 열린 파일을 유지하면 많은 메모리를 소비할 수 있어요. 체크포인트 시점에 150개의 파일(아무리 작아도)을 플러시하고 업로드하는 것도 느릴 수 있어요. 이 구성이 0.02로 설정되면, 모든 쓰기 작업에 대해 각 정렬 키가 목표 가중치 1,000의 2%인 기본 가중치를 가진다는 뜻이에요. 이렇게 하면 아무리 작아도 하나의 작성자 작업에 50개 이상의 데이터 파일(하루에 하나)을 배치하지 않게 돼요.

이것은 저카디널리티 시나리오의 StatisticsType.Map에만 적용돼요. StatisticsType.Sketch 고카디널리티 정렬 컬럼의 경우 보통 파티션 컬럼으로 사용되지 않아요. 그렇지 않으면 쓰기 중 너무 많은 파티션과 작은 파일이 생성될 수 있기 때문이에요. Sketch 범위 파티셔너는 간단히 고카디널리티 키를 정렬된 범위로 나눠요.

더 알아보기 (Learn more)

• 플링크 사용 시작 (Flink getting started)
• 카탈로그 속성 (Catalog properties)
• 플링크 DDL