플링크 쿼리

플링크 쿼리 (Flink Queries)

아이스버그는 아파치 플링크의 DataStream API와 Table API로 스트리밍 및 배치 읽기를 지원해요. 이 문서에서는 Flink SQL과 DataStream API로 아이스버그 테이블을 읽는 방법, FLIP-27 소스 사용, 브랜치·태그 읽기, 워터마크 생성, 그리고 아이스버그의 다양한 메타데이터 테이블을 활용한 테이블 검사 방법을 알려드릴게요.

출처: 문서

본문

아이스버그는 아파치 플링크의 DataStream API와 Table API로 스트리밍 및 배치 읽기를 지원해요.

SQL로 읽기 (Reading with SQL)

아이스버그는 Flink에서 스트리밍과 배치 읽기를 모두 지원해요. 다음 SQL 명령을 실행해서 실행 모드를 스트리밍에서 배치로, 또는 그 반대로 전환해요.

-- Execute the flink job in streaming mode for current session context
SET execution.runtime-mode = streaming;

-- Execute the flink job in batch mode for current session context
SET execution.runtime-mode = batch;

Flink 배치 읽기 (Flink batch read)

다음 문장으로 Flink 배치 작업을 제출해요.

-- Execute the flink job in batch mode for current session context
SET execution.runtime-mode = batch;
SELECT * FROM sample;

Flink 스트리밍 읽기 (Flink streaming read)

아이스버그는 과거 스냅샷 ID에서 시작하는 Flink 스트리밍 작업에서 증분 데이터를 처리하는 것을 지원해요.

-- Submit the flink job in streaming mode for current session.
SET execution.runtime-mode = streaming;

-- Enable this switch because streaming read SQL will provide few job options in flink SQL hint options.
SET table.dynamic-table-options.enabled=true;

-- Read all the records from the iceberg current snapshot, and then read incremental data starting from that snapshot.
SELECT * FROM sample /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s')*/ ;

-- Read all incremental data starting from the snapshot-id '3821550127947089987' (records from this snapshot will be excluded).
SELECT * FROM sample /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s', 'start-snapshot-id'='3821550127947089987')*/ ;

스트리밍 작업을 위해 Flink SQL 힌트 옵션에 설정할 수 있는 옵션이 몇 가지 있어요. 자세한 내용은 읽기 옵션(read options)을 참고해주세요.

SQL용 FLIP-27 소스 (FLIP-27 source for SQL)

다음은 FLIP-27 소스에 옵트인하거나 옵트아웃하는 SQL 설정이에요.

-- Opt out the FLIP-27 source.
-- Default is false for Flink 1.19 and below, and true for Flink 1.20 and above.
SET table.exec.iceberg.use-flip27-source = false;

위에서 문서화한 다른 모든 SQL 설정과 옵션은 FLIP-27 소스에 적용돼요.

SQL로 브랜치와 태그 읽기 (Reading branches and tags with SQL)

브랜치와 태그는 옵션을 지정해서 SQL로 읽을 수 있어요. 자세한 내용은 Flink 구성(Flink Configuration)을 참고해주세요.

--- Read from branch b1
SELECT * FROM table /*+ OPTIONS('branch'='b1') */ ;

--- Read from tag t1
SELECT * FROM table /*+ OPTIONS('tag'='t1') */;

--- Incremental scan from tag t1 to tag t2
SELECT * FROM table /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s', 'start-tag'='t1', 'end-tag'='t2') */;

DataStream으로 읽기 (Reading with DataStream)

아이스버그는 이제 자바 API에서 스트리밍 또는 배치 읽기를 지원해요.

배치 읽기 (Batch Read)

이 예시는 아이스버그 테이블의 모든 레코드를 읽은 다음 Flink 배치 작업에서 stdout 콘솔로 출력해요.

StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://nn:8020/warehouse/path");
DataStream<RowData> batch = FlinkSource.forRowData()
     .env(env)
     .tableLoader(tableLoader)
     .streaming(false)
     .build();

// Print all records to stdout.
batch.print();

// Submit and execute this batch read job.
env.execute("Test Iceberg Batch Read");

스트리밍 읽기 (Streaming read)

이 예시는 snapshot-id '3821550127947089987'에서 시작하는 증분 레코드를 읽고 Flink 스트리밍 작업에서 stdout 콘솔로 출력해요.

StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://nn:8020/warehouse/path");
DataStream<RowData> stream = FlinkSource.forRowData()
     .env(env)
     .tableLoader(tableLoader)
     .streaming(true)
     .startSnapshotId(3821550127947089987L)
     .build();

// Print all records to stdout.
stream.print();

// Submit and execute this streaming read job.
env.execute("Test Iceberg Streaming Read");

설정할 수 있는 다른 옵션도 있어요. FlinkSource#Builder를 참고해주세요.

DataStream으로 읽기 (FLIP-27 소스) (Reading with DataStream (FLIP-27 source))

FLIP-27 소스 인터페이스는 Flink 1.12에서 도입됐어요. 이는 이전 SourceFunction 스트리밍 소스 인터페이스의 여러 단점을 해결하는 것을 목표로 해요. 또한 배치와 스트리밍 실행 모두를 위해 소스 인터페이스를 통합해요. Flink 저장소의 대부분 소스 커넥터(Kafka, file 등)는 FLIP-27 인터페이스로 마이그레이션했어요. Flink는 가까운 시일 내에 이전 SourceFunction 인터페이스를 더 이상 사용하지 않게(deprecate) 할 계획이에요.

FLIP-27 기반 Flink IcebergSource가 iceberg-flink 모듈에 추가됐어요. FLIP-27 IcebergSource는 현재 실험적 기능이에요.

배치 읽기 (Batch Read)

이 예시는 아이스버그 테이블의 모든 레코드를 읽은 다음 Flink 배치 작업에서 stdout 콘솔로 출력해요.

StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://nn:8020/warehouse/path");

IcebergSource<RowData> source = IcebergSource.forRowData()
    .tableLoader(tableLoader)
    .assignerFactory(new SimpleSplitAssignerFactory())
    .build();

DataStream<RowData> batch = env.fromSource(
    source,
    WatermarkStrategy.noWatermarks(),
    "My Iceberg Source",
    TypeInformation.of(RowData.class));

// Print all records to stdout.
batch.print();

// Submit and execute this batch read job.
env.execute("Test Iceberg Batch Read");

스트리밍 읽기 (Streaming read)

이 예시는 최신 테이블 스냅샷(포함)부터 스트리밍 읽기를 시작해요. 60초마다 아이스버그 테이블을 폴링해서 새 append-only 스냅샷을 발견해요. CDC 읽기는 아직 지원되지 않아요.

StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://nn:8020/warehouse/path");

IcebergSource source = IcebergSource.forRowData()
    .tableLoader(tableLoader)
    .assignerFactory(new SimpleSplitAssignerFactory())
    .streaming(true)
    .streamingStartingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT)
    .monitorInterval(Duration.ofSeconds(60))
    .build();

DataStream<RowData> stream = env.fromSource(
    source,
    WatermarkStrategy.noWatermarks(),
    "My Iceberg Source",
    TypeInformation.of(RowData.class));

// Print all records to stdout.
stream.print();

// Submit and execute this streaming read job.
env.execute("Test Iceberg Streaming Read");

Java API로 설정할 수 있는 다른 옵션도 있어요. IcebergSource#Builder를 참고해주세요.

DataStream으로 브랜치와 태그 읽기 (Reading branches and tags with DataStream)

브랜치와 태그는 DataStream API로도 읽을 수 있어요.

StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://nn:8020/warehouse/path");
// Read from branch
DataStream<RowData> batch = FlinkSource.forRowData()
    .env(env)
    .tableLoader(tableLoader)
    .branch("test-branch")
    .streaming(false)
    .build();

// Read from tag
DataStream<RowData> batch = FlinkSource.forRowData()
    .env(env)
    .tableLoader(tableLoader)
    .tag("test-tag")
    .streaming(false)
    .build();

// Streaming read from start-tag
DataStream<RowData> batch = FlinkSource.forRowData()
    .env(env)
    .tableLoader(tableLoader)
    .streaming(true)
    .startTag("test-tag")
    .build();

Avro GenericRecord로 읽기 (Read as Avro GenericRecord)

FLIP-27 아이스버그 소스는 Flink RowData를 Avro GenericRecord로 변환하는 AvroGenericRecordReaderFunction을 제공해요. 이 변환을 사용해서 아이스버그 테이블을 Avro GenericRecord DataStream으로 읽을 수 있어요.

flink-avro jar가 클래스패스에 포함돼 있는지 확인해주세요. 또한 iceberg-flink-runtime shaded bundle jar는 런타임 jar가 avro 패키지를 쉐이딩하기 때문에 사용할 수 없어요. 대신 non-shaded iceberg-flink jar를 사용해주세요.

TableLoader tableLoader = ...;
Table table;
try (TableLoader loader = tableLoader) {
    loader.open();
    table = loader.loadTable();
}

AvroGenericRecordReaderFunction readerFunction = AvroGenericRecordReaderFunction.fromTable(table);

IcebergSource<GenericRecord> source =
    IcebergSource.<GenericRecord>builder()
        .tableLoader(tableLoader)
        .readerFunction(readerFunction)
        .assignerFactory(new SimpleSplitAssignerFactory())
        ...
        .build();

DataStream<Row> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(),
    "Iceberg Source as Avro GenericRecord", new GenericRecordAvroTypeInfo(avroSchema));

워터마크 생성 (Emitting watermarks)

소스 자체에서 워터마크를 생성하는 것은 여러 목적에 유용할 수 있어요. 예를 들어 Flink Watermark Alignment를 활용하거나, 여러 데이터 파일을 동시에 읽을 때 윈도우가 너무 일찍 트리거되는 것을 방지할 수 있어요.

watermarkColumn을 설정해서 IcebergSource의 워터마크 생성을 활성화해요. 지원되는 컬럼 타입은 timestamp, timestamptz, long이에요. 아이스버그 timestamp 또는 timestamptz는 본질적으로 시간 정밀도를 포함해요. 그래서 시간 단위를 지정할 필요가 없어요. 하지만 long 타입 컬럼은 시간 단위 정보를 포함하지 않아요. long 컬럼의 변환을 구성하려면 watermarkTimeUnit을 사용해요.

워터마크는 데이터 파일에 저장된 컬럼 메트릭을 기반으로 생성되고 스플릿당 한 번 생성돼요. 서로 다른 시간 범위의 여러 작은 파일이 단일 스플릿으로 결합되면, 무질서도(out-of-orderliness)와 Flink 상태의 추가 데이터 버퍼링이 증가할 수 있어요. 워터마크 정렬의 주요 목적은 Flink 상태의 무질서도와 과도한 데이터 버퍼링을 줄이는 것이에요. 따라서 여러 작은 파일이 단일 스플릿으로 결합되는 것을 방지하려면 read.split.open-file-cost를 매우 큰 값으로 설정하는 것을 권장해요. (작은 파일을 단일 스플릿으로 결합하지 않는 것의) 부정적 영향은 읽기 처리량에 있어요. 특히 작은 파일이 많을 때 그렇죠. 전형적인 상태 저장 프로세싱 작업에서 소스 읽기 처리량은 병목이 아니에요. 그러니 이는 합리적인 트레이드오프일 가능성이 높아요.

이 기능은 컬럼 수준 min-max 통계가 필요해요. 쓰기 단계에서 워터마크 컬럼에 대한 통계가 생성되는지 확인해주세요. 기본적으로 컬럼 메트릭은 테이블의 처음 100개 컬럼에 대해 수집돼요. 워터마크 컬럼이 기본으로 통계를 활성화하지 않는다면, 필요할 때 write.metadata.metrics로 시작하는 쓰기 속성을 사용해요.

다음 예시는 워터마크를 윈도우에 사용할 때 유용할 수 있어요. 소스는 타임스탬프 컬럼을 사용해 아이스버그 데이터 파일을 순서대로 읽고 워터마크를 생성해요.

StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://nn:8020/warehouse/path");

DataStream<RowData> stream =
    env.fromSource(
        IcebergSource.forRowData()
            .tableLoader(tableLoader)
            // Watermark using timestamp column
            .watermarkColumn("timestamp_column")
            .build(),
        // Watermarks are generated by the source, no need to generate it manually
        WatermarkStrategy.<RowData>noWatermarks()
            // Extract event timestamp from records
            .withTimestampAssigner((record, eventTime) -> record.getTimestamp(pos, precision).getMillisecond()),
        SOURCE_NAME,
        TypeInformation.of(RowData.class));

워터마크 정렬에 long 이벤트 컬럼을 사용해 아이스버그 테이블을 읽는 예시:

StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://nn:8020/warehouse/path");

DataStream<RowData> stream =
    env.fromSource(
        IcebergSource source = IcebergSource.forRowData()
            .tableLoader(tableLoader)
            // Disable combining multiple files to a single split
            .set(FlinkReadOptions.SPLIT_FILE_OPEN_COST, String.valueOf(TableProperties.SPLIT_SIZE_DEFAULT))
            // Watermark using long column
            .watermarkColumn("long_column")
            .watermarkTimeUnit(TimeUnit.MILLI_SCALE)
            .build(),
        // Watermarks are generated by the source, no need to generate it manually
        WatermarkStrategy.<RowData>noWatermarks()
            .withWatermarkAlignment(watermarkGroup, maxAllowedWatermarkDrift),
        SOURCE_NAME,
        TypeInformation.of(RowData.class));

옵션 (Options)

읽기 옵션 (Read options)

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

IcebergSource.forRowData()
    .tableLoader(TableLoader.fromCatalog(...))
    .assignerFactory(new SimpleSplitAssignerFactory())
    .streaming(true)
    .streamingStartingStrategy(StreamingStartingStrategy.INCREMENTAL_FROM_LATEST_SNAPSHOT)
    .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);
...

모든 옵션은 여기(read-options)에서 확인할 수 있어요.

테이블 검사 (Inspecting tables)

테이블의 이력, 스냅샷 및 기타 메타데이터를 검사하기 위해 아이스버그는 메타데이터 테이블을 지원해요.

메타데이터 테이블은 원래 테이블 이름 뒤에 메타데이터 테이블 이름을 붙여서 식별해요. 예를 들어 db.table의 이력은 db.table$history로 읽어요.

이력 (History)

테이블 이력을 보려면:

SELECT * FROM prod.db.table$history;

made_current_at	snapshot_id	parent_id	is_current_ancestor
2019-02-08 03:29:51.215	5781947118336215154	NULL	true
2019-02-08 03:47:55.948	5179299526185056830	5781947118336215154	true
2019-02-09 16:24:30.13	296410040247533544	5179299526185056830	false
2019-02-09 16:32:47.336	2999875608062437330	5179299526185056830	true
2019-02-09 19:42:03.919	8924558786060583479	2999875608062437330	true
2019-02-09 19:49:16.343	6536733823181975045	8924558786060583479	true

정보 (Info)

이것은 롤백된 커밋을 보여줘요. 이 예시에서 스냅샷 296410040247533544와 2999875608062437330은 같은 부모 스냅샷 5179299526185056830을 가져요. 스냅샷 296410040247533544는 롤백되었고 현재 테이블 상태의 조상이 아니에요.

메타데이터 로그 항목 (Metadata Log Entries)

테이블 메타데이터 로그 항목을 보려면:

SELECT * from prod.db.table$metadata_log_entries;

timestamp	file	latest_snapshot_id	latest_schema_id	latest_sequence_number
2022-07-28 10:43:52.93	s3://.../table/metadata/00000-9441e604-b3c2-498a-a45a-6320e8ab9006.metadata.json	null	null	null
2022-07-28 10:43:57.487	s3://.../table/metadata/00001-f30823df-b745-4a0a-b293-7532e0c99986.metadata.json	170260833677645300	0	1
2022-07-28 10:43:58.25	s3://.../table/metadata/00002-2cc2837a-02dc-4687-acc1-b4d86ea486f4.metadata.json	958906493976709774	0	2

스냅샷 (Snapshots)

테이블의 유효한 스냅샷을 보려면:

SELECT * FROM prod.db.table$snapshots;

committed_at	snapshot_id	parent_id	operation	manifest_list	summary
2019-02-08 03:29:51.215	57897183625154	null	append	s3://.../table/metadata/snap-57897183625154-1.avro	{ added-records -> 2478404, total-records -> 2478404, added-data-files -> 438, total-data-files -> 438, flink.job-id -> 2e274eecb503d85369fb390e8956c813 }

스냅샷을 테이블 이력과 조인할 수도 있어요. 예를 들어 이 쿼리는 각 스냅샷을 쓴 애플리케이션 ID와 함께 테이블 이력을 보여줘요.

select
    h.made_current_at,
    s.operation,
    h.snapshot_id,
    h.is_current_ancestor,
    s.summary['flink.job-id']
from prod.db.table$history h
join prod.db.table$snapshots s
  on h.snapshot_id = s.snapshot_id
order by made_current_at;

made_current_at	operation	snapshot_id	is_current_ancestor	summary[flink.job-id]
2019-02-08 03:29:51.215	append	57897183625154	true	2e274eecb503d85369fb390e8956c813

파일 (Files)

테이블의 현재 데이터 파일을 보려면:

SELECT * FROM prod.db.table$files;

content	file_path	file_format	spec_id	partition	record_count	file_size_in_bytes	column_sizes	value_counts	null_value_counts	nan_value_counts	lower_bounds	upper_bounds	key_metadata	split_offsets	equality_ids	sort_order_id
0	s3:/.../table/data/00000-3-8d6d60e8-d427-4809-bcf0-f5d45a4aad96.parquet	PARQUET	0	{1999-01-01, 01}	1	597	[1 -> 90, 2 -> 62]	[1 -> 1, 2 -> 1]	[1 -> 0, 2 -> 0]	[]	[1 -> , 2 -> c]	[1 -> , 2 -> c]	null	[4]	null	null
0	s3:/.../table/data/00001-4-8d6d60e8-d427-4809-bcf0-f5d45a4aad96.parquet	PARQUET	0	{1999-01-01, 02}	1	597	[1 -> 90, 2 -> 62]	[1 -> 1, 2 -> 1]	[1 -> 0, 2 -> 0]	[]	[1 -> , 2 -> b]	[1 -> , 2 -> b]	null	[4]	null	null
0	s3:/.../table/data/00002-5-8d6d60e8-d427-4809-bcf0-f5d45a4aad96.parquet	PARQUET	0	{1999-01-01, 03}	1	597	[1 -> 90, 2 -> 62]	[1 -> 1, 2 -> 1]	[1 -> 0, 2 -> 0]	[]	[1 -> , 2 -> a]	[1 -> , 2 -> a]	null	[4]	null	null

매니페스트 (Manifests)

테이블의 현재 파일 매니페스트를 보려면:

SELECT * FROM prod.db.table$manifests;

path	length	partition_spec_id	added_snapshot_id	added_data_files_count	existing_data_files_count	deleted_data_files_count	partition_summaries
s3://.../table/metadata/45b5290b-ee61-4788-b324-b1e2735c0e10-m0.avro	4479	0	6668963634911763636	8	0	0	[[false,null,2019-05-13,2019-05-15]]

참고:

1. manifests 테이블의 partition_summaries 컬럼 안의 필드는 매니페스트 리스트 안의 field_summary struct에 해당하며, 순서는 contains_null contains_nan lower_bound upper_bound예요.
2. contains_nan은 null을 반환할 수 있는데, 이는 파일 메타데이터에서 이 정보를 사용할 수 없다는 뜻이에요. 이는 보통 contains_nan이 채워지지 않는 V1 테이블에서 읽을 때 발생해요.

파티션 (Partitions)

테이블의 현재 파티션을 보려면:

SELECT * FROM prod.db.table$partitions;

partition	spec_id	record_count	file_count	total_data_file_size_in_bytes	position_delete_record_count	position_delete_file_count	equality_delete_record_count	equality_delete_file_count	last_updated_at(μs)	last_updated_snapshot_id
{20211001, 11}	0	1	1	100	2	1	0	0	1633086034192000	9205185327307503337
{20211002, 11}	0	4	3	500	1	1	0	0	1633172537358000	867027598972211003
{20211001, 10}	0	7	4	700	0	0	0	0	1633082598716000	3280122546965981531
{20211002, 10}	0	3	2	400	0	0	1	1	1633169159489000	6941468797545315876

참고: 파티셔닝되지 않은 테이블의 경우 partitions 테이블에는 partition과 spec_id 필드가 없어요.

모든 메타데이터 테이블 (All Metadata Tables)

이 테이블들은 현재 스냅샷에 특화된 메타데이터 테이블들의 합집합이고, 모든 스냅샷에 걸친 메타데이터를 반환해요.

위험 (Danger)

"all" 메타데이터 테이블은 메타데이터 파일이 둘 이상의 테이블 스냅샷에 속할 수 있으므로 데이터 파일이나 매니페스트 파일당 한 행 이상을 만들 수 있어요.

모든 데이터 파일 (All Data Files)

테이블의 모든 데이터 파일과 각 파일의 메타데이터를 보려면:

SELECT * FROM prod.db.table$all_data_files;

content	file_path	file_format	partition	record_count	file_size_in_bytes	column_sizes	value_counts	null_value_counts	nan_value_counts	lower_bounds	upper_bounds	key_metadata	split_offsets	equality_ids	sort_order_id
0	s3://.../dt=20210102/00000-0-756e2512-49ae-45bb-aae3-c0ca475e7879-00001.parquet	PARQUET	{20210102}	14	2444	{1 -> 94, 2 -> 17}	{1 -> 14, 2 -> 14}	{1 -> 0, 2 -> 0}	{}	{1 -> 1, 2 -> 20210102}	{1 -> 2, 2 -> 20210102}	null	[4]	null	0
0	s3://.../dt=20210103/00000-0-26222098-032f-472b-8ea5-651a55b21210-00001.parquet	PARQUET	{20210103}	14	2444	{1 -> 94, 2 -> 17}	{1 -> 14, 2 -> 14}	{1 -> 0, 2 -> 0}	{}	{1 -> 1, 2 -> 20210103}	{1 -> 3, 2 -> 20210103}	null	[4]	null	0
0	s3://.../dt=20210104/00000-0-a3bb1927-88eb-4f1c-bc6e-19076b0d952e-00001.parquet	PARQUET	{20210104}	14	2444	{1 -> 94, 2 -> 17}	{1 -> 14, 2 -> 14}	{1 -> 0, 2 -> 0}	{}	{1 -> 1, 2 -> 20210104}	{1 -> 3, 2 -> 20210104}	null	[4]	null	0

모든 매니페스트 (All Manifests)

테이블의 모든 매니페스트 파일을 보려면:

SELECT * FROM prod.db.table$all_manifests;

path	length	partition_spec_id	added_snapshot_id	added_data_files_count	existing_data_files_count	deleted_data_files_count	partition_summaries
s3://.../metadata/a85f78c5-3222-4b37-b7e4-faf944425d48-m0.avro	6376	0	6272782676904868561	2	0	0	[{false, false, 20210101, 20210101}]

참고:

1. manifests 테이블의 partition_summaries 컬럼 안의 필드는 매니페스트 리스트 안의 field_summary struct에 해당하며, 순서는 contains_null contains_nan lower_bound upper_bound예요.
2. contains_nan은 null을 반환할 수 있는데, 이는 파일 메타데이터에서 이 정보를 사용할 수 없다는 뜻이에요. 이는 보통 contains_nan이 채워지지 않는 V1 테이블에서 읽을 때 발생해요.

참조 (References)

테이블의 알려진 스냅샷 참조를 보려면:

SELECT * FROM prod.db.table$refs;

name	type	snapshot_id	max_reference_age_in_ms	min_snapshots_to_keep	max_snapshot_age_in_ms
main	BRANCH	4686954189838128572	10	20	30
testTag	TAG	4686954189838128572	10	null	null

더 알아보기 (Learn more)

• 플링크 사용 시작 (Flink getting started)
• 플링크 구성 (Flink configuration)
• 플링크 쓰기 (Flink writes)