플링크 DDL

플링크 DDL

이 문서에서는 Flink SQL에서 아이스버그를 사용할 때 쓸 수 있는 DDL 명령들을 알려드릴게요. CREATE CATALOG로 Hive, Hadoop, REST, 커스텀 카탈로그를 만드는 방법부터 CREATE DATABASE, CREATE TABLE, CREATE TABLE LIKE, ALTER TABLE, DROP TABLE까지 기본적인 DDL 사용법을 실제 SQL 예시와 함께 살펴볼게요.

출처: 문서

본문

DDL 명령 (DDL commands)

CREATE CATALOG

Hive 카탈로그 (Hive catalog)

'catalog-type'='hive'로 구성할 수 있고 Hive metastore에서 테이블을 로드하는 hive_catalog라는 아이스버그 카탈로그를 만들어요.

CREATE CATALOG hive_catalog WITH (
  'type'='iceberg',
  'catalog-type'='hive',
  'uri'='thrift://localhost:9083',
  'clients'='5',
  'property-version'='1',
  'warehouse'='hdfs://nn:8020/warehouse/path'
);

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

• uri: Hive metastore의 thrift URI. (필수)
• clients: Hive metastore 클라이언트 풀 크기, 기본값은 2. (선택)
• warehouse: Hive warehouse 위치. hive-conf-dir을 설정해서 hive-site.xml 구성 파일이 들어있는 위치를 지정하지 않았거나 올바른 hive-site.xml을 classpath에 추가하지 않았다면 사용자가 이 경로를 지정해야 해요.
• hive-conf-dir: 커스텀 Hive 구성 값을 제공하는 데 사용될 hive-site.xml 구성 파일이 들어있는 디렉터리의 경로. 아이스버그 카탈로그를 만들 때 hive-conf-dir와 warehouse를 모두 설정하면 /hive-site.xml(또는 classpath의 hive 구성 파일)의 hive.metastore.warehouse.dir 값이 warehouse 값으로 덮어써져요.
• hadoop-conf-dir: 커스텀 Hadoop 구성 값을 제공하는 데 사용될 core-site.xml과 hdfs-site.xml 구성 파일이 들어있는 디렉터리의 경로.

Hive 카탈로그 제한 (Hive Catalog Limitation)

Hive Metastore(HMS)는 컬럼 타입을 위치적으로 비교해서 스키마 변경을 검증해요(hive.metastore.disallow.incompatible.col.type.changes, 기본값 true). Hive 카탈로그를 사용할 때, 마지막이 아닌 컬럼을 드롭하거나 컬럼을 재정렬하는 것처럼 컬럼 위치를 바꾸는 스키마 진화 연산은 어떤 엔진이 수행하든(Spark, Flink Java API 등) 실패할 수 있어요.

이를 우회하려면 hive.metastore.disallow.incompatible.col.type.changes=false로 설정해서 HMS 스키마 호환성 검사를 비활성화해요:

• 원격 HMS: HMS 서버의 hive-site.xml에 이 속성을 설정해요.
• 임베디드 HMS: Hive 카탈로그 구성에 해당 속성을 추가해요.

트레이드오프: 이 검사를 비활성화한 후에는 Hive Metastore의 스키마 불일치로 인해 Hive 엔진이 테이블을 올바르게 읽지 못할 수 있어요. 아이스버그를 아는 엔진(Spark, Flink, Trino 등)은 Hive Metastore가 아니라 아이스버그 메타데이터에서 스키마를 읽으므로 계속 올바르게 동작해요.

Hadoop 카탈로그 (Hadoop catalog)

아이스버그는 'catalog-type'='hadoop'으로 구성할 수 있는 HDFS의 디렉터리 기반 카탈로그도 지원해요.

CREATE CATALOG hadoop_catalog WITH (
  'type'='iceberg',
  'catalog-type'='hadoop',
  'warehouse'='hdfs://nn:8020/warehouse/path',
  'property-version'='1'
);

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

• warehouse: 메타데이터 파일과 데이터 파일을 저장할 HDFS 디렉터리. (필수)

USE CATALOG hadoop_catalog SQL 명령을 실행해서 현재 카탈로그를 설정해요.

REST 카탈로그 (REST catalog)

'catalog-type'='rest'로 구성할 수 있고 REST 카탈로그에서 테이블을 로드하는 rest_catalog라는 아이스버그 카탈로그를 만들어요.

CREATE CATALOG rest_catalog WITH (
  'type'='iceberg',
  'catalog-type'='rest',
  'uri'='https://localhost/'
);

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

• uri: REST 카탈로그의 URL (필수)
• credential: OAuth2 클라이언트 자격증명 흐름에서 토큰과 교환할 자격증명 (선택)
• token: 서버와 상호작용하는 데 사용될 토큰 (선택)

커스텀 카탈로그 (Custom catalog)

Flink는 catalog-impl 속성을 지정해서 커스텀 아이스버그 카탈로그 구현을 로드하는 것도 지원해요.

CREATE CATALOG my_catalog WITH (
  'type'='iceberg',
  'catalog-impl'='com.my.custom.CatalogImpl',
  'my-additional-catalog-config'='my-value'
);

YAML 구성을 통한 생성 (Create through YAML config)

SQL 클라이언트를 시작하기 전에 카탈로그를 sql-client-defaults.yaml에 등록할 수 있어요.

catalogs: 
  - name: my_catalog
    type: iceberg
    catalog-type: hadoop
    warehouse: hdfs://nn:8020/warehouse/path

SQL 파일을 통한 생성 (Create through SQL Files)

Flink SQL 클라이언트는 -i 시작 옵션을 지원해서 SQL 클라이언트를 시작할 때 환경을 설정하는 초기화 SQL 파일을 실행해요.

-- define available catalogs
CREATE CATALOG hive_catalog WITH (
  'type'='iceberg',
  'catalog-type'='hive',
  'uri'='thrift://localhost:9083',
  'warehouse'='hdfs://nn:8020/warehouse/path'
);

USE CATALOG hive_catalog;

-i <init.sql> 옵션을 사용해서 SQL 클라이언트 세션을 초기화해요.

/path/to/bin/sql-client.sh -i /path/to/init.sql

CREATE DATABASE

기본적으로 아이스버그는 Flink의 기본 데이터베이스를 사용해요. 기본 데이터베이스 아래에 테이블을 만드는 것을 피하기 위해 다음 예시처럼 별도의 데이터베이스를 만들어요.

CREATE DATABASE iceberg_db;
USE iceberg_db;

CREATE TABLE

CREATE TABLE `hive_catalog`.`default`.`sample` (
    id BIGINT COMMENT 'unique id',
    data STRING NOT NULL
) WITH ('format-version'='2');

테이블 생성 명령은 자주 사용되는 Flink create 절을 지원해요:

• PARTITION BY (column1, column2, ...)로 파티셔닝 구성. Flink는 아직 숨겨진 파티셔닝(hidden partitioning)을 지원하지 않아요.
• COMMENT 'table document'로 테이블 설명 설정.
• WITH ('key'='value', ...)로 Iceberg 테이블 속성에 저장될 테이블 구성을 설정.

테이블 위치를 지정하려면 WITH ('location'='fully-qualified-uri')를 사용해요:

CREATE TABLE `hive_catalog`.`default`.`sample` (
    id BIGINT COMMENT 'unique id',
    data STRING NOT NULL
) WITH (
    'format-version'='2', 
    'location'='hdfs//nn:8020/custom-path'
);

현재 계산된 컬럼(computed column)과 워터마크 정의 등은 지원하지 않아요.

PRIMARY KEY

기본 키 제약은 하나의 컬럼이나 컬럼 집합에 선언할 수 있고, 고유해야 하며 null을 포함하지 않아야 해요. UPSERT 모드에 필요해요.

CREATE TABLE `hive_catalog`.`default`.`sample` (
    id BIGINT COMMENT 'unique id',
    data STRING NOT NULL,
    PRIMARY KEY(`id`) NOT ENFORCED
) WITH ('format-version'='2');

PARTITIONED BY

파티션 테이블을 만들려면 PARTITIONED BY를 사용해요:

CREATE TABLE `hive_catalog`.`default`.`sample` (
    id BIGINT COMMENT 'unique id',
    data STRING NOT NULL
) 
PARTITIONED BY (data) 
WITH ('format-version'='2');

아이스버그는 숨겨진 파티셔닝을 지원하지만 Flink는 컬럼에 대한 함수로 파티셔닝하는 것을 지원하지 않아요. Flink DDL에서는 숨겨진 파티션을 지원할 방법이 없어요.

CREATE TABLE LIKE

다른 테이블과 같은 스키마, 파티셔닝, 테이블 속성을 가진 테이블을 만들려면 CREATE TABLE LIKE를 사용해요.

CREATE TABLE `hive_catalog`.`default`.`sample` (
    id BIGINT COMMENT 'unique id',
    data STRING
);

CREATE TABLE  `hive_catalog`.`default`.`sample_like` LIKE `hive_catalog`.`default`.`sample`;

자세한 내용은 Flink CREATE TABLE 문서를 참고해주세요.

ALTER TABLE

아이스버그는 테이블 속성 변경만 지원해요:

ALTER TABLE `hive_catalog`.`default`.`sample` SET ('write.format.default'='avro');

ALTER TABLE .. RENAME TO

ALTER TABLE `hive_catalog`.`default`.`sample` RENAME TO `hive_catalog`.`default`.`new_sample`;

DROP TABLE

테이블을 삭제하려면 다음을 실행해요:

DROP TABLE `hive_catalog`.`default`.`sample`;

더 알아보기 (Learn more)

• 플링크 사용 시작 (Flink getting started)
• 플링크 쿼리 (Flink queries)
• 플링크 쓰기 (Flink writes)