Apache Iceberg 성능 튜닝 가이드 — 주의 사항과 핵심 파라미터 총정리

ALTER TABLE db.orders SET TBLPROPERTIES (
  'write.target-file-size-bytes' = '536870912',
  'write.parquet.compression-codec' = 'zstd',
  'write.parquet.compression-level' = '3',
  'write.metadata.compression-codec' = 'gzip',
  'write.object-storage.enabled' = 'true'
);

ALTER TABLE db.orders SET TBLPROPERTIES (
  'write.metadata.metrics.default' = 'counts',
  'write.metadata.metrics.column.order_id' = 'full',
  'write.metadata.metrics.column.ts' = 'full',
  'write.metadata.metrics.column.customer_id' = 'full'
);

ALTER TABLE db.orders SET TBLPROPERTIES (
  'read.split.target-size' = '268435456',
  'read.split.open-file-cost' = '8388608'
);

CALL demo.system.rewrite_data_files(
  table => 'db.orders',
  strategy => 'sort',
  sort_order => 'ts ASC NULLS LAST, customer_id ASC',
  options => map(
    'target-file-size-bytes', '536870912',
    'max-concurrent-file-group-rewrites', '10',
    'partial-progress.enabled', 'true',
    'partial-progress.max-commits', '50',
    'rewrite-job-order', 'bytes-asc'
  )
);

CALL demo.system.rewrite_data_files(
  table => 'db.orders',
  strategy => 'sort',
  sort_order => 'zorder(customer_id, ts)',
  options => map('target-file-size-bytes', '536870912')
);

CALL demo.system.rewrite_position_deletes(
  table => 'db.orders',
  options => map(
    'rewrite-job-order', 'bytes-asc',
    'partial-progress.enabled', 'true'
  )
);

CALL demo.system.rewrite_manifests(
  table => 'db.orders',
  use_caching => true
);

# Iceberg vectorized reader
spark.sql.iceberg.vectorization.enabled = true
 
# Timestamp 처리 (timezone-aware)
spark.sql.iceberg.handle-timestamp-without-timezone = false
 
# Catalog 캐시
spark.sql.catalog.demo.cache-enabled       = true
spark.sql.catalog.demo.cache.expiration-interval-ms = 30000
 
# Adaptive Query Execution
spark.sql.adaptive.enabled                       = true
spark.sql.adaptive.coalescePartitions.enabled    = true
spark.sql.adaptive.skewJoin.enabled              = true
 
# Shuffle partitions (write distribution=hash와 함께)
spark.sql.shuffle.partitions = 800

(df.writeTo("demo.db.orders")
    .option("write-format", "parquet")
    .option("target-file-size-bytes", "536870912")
    .option("check-ordering", "false")     # 정렬 검증 스킵
    .option("fanout-enabled", "true")      # 파티션이 많을 때
    .append())

(spark.read.format("iceberg")
    .option("split-size", "268435456")
    .option("lookback", "20")
    .option("file-open-cost", "8388608")
    .option("vectorization-enabled", "true")
    .load("demo.db.orders"))

connector.name = iceberg
iceberg.catalog.type = rest
iceberg.rest-catalog.uri = http://iceberg-rest:8181
 
# 성능 관련
iceberg.split-size                          = 128MB
iceberg.max-splits-per-second               = 100
iceberg.use-file-size-from-metadata         = true
 
# Manifest 캐시 (중요)
iceberg.metadata-cache.enabled              = true
iceberg.metadata-cache.max-size             = 1000
iceberg.metadata-cache.ttl                  = 1h
iceberg.io.manifest.cache-enabled           = true
iceberg.io.manifest.cache.max-total-bytes   = 104857600
iceberg.io.manifest.cache.expiration-interval-ms = 60000
 
# Parquet
parquet.use-column-index                    = true
parquet.use-bloom-filter                    = true
parquet.max-read-block-size                 = 16MB
 
# Statistics 기반 CBO
iceberg.table-statistics-enabled            = true
iceberg.extended-statistics.enabled         = true

CREATE TABLE orders_iceberg (
  id BIGINT,
  ts TIMESTAMP(3),
  amount DECIMAL(10,2)
) WITH (
  'connector' = 'iceberg',
  'catalog-type' = 'rest',
  'uri' = 'http://iceberg-rest:8181',
  'warehouse' = 's3://my-warehouse',
 
  -- Commit 빈도 (가장 중요)
  'write-parallelism' = '8',
  'upsert-enabled' = 'true',
  'equality-field-columns' = 'id',
 
  -- 파일 크기
  'write.target-file-size-bytes' = '134217728',   -- 128MB (스트리밍은 작게)
 
  -- Checkpoint 기반 commit
  'write.distribution-mode' = 'hash'
);

# flink-conf.yaml
execution.checkpointing.interval = 5min
execution.checkpointing.min-pause = 1min

ALTER TABLE db.orders SET TBLPROPERTIES (
  'write.target-file-size-bytes' = '1073741824',   -- 1GB
  'write.parquet.compression-codec' = 'zstd',
  'write.parquet.compression-level' = '6',
  'write.parquet.row-group-size-bytes' = '268435456',
  'write.delete.mode' = 'copy-on-write',
  'write.distribution-mode' = 'range',
  'read.split.target-size' = '536870912'
);

ALTER TABLE db.events SET TBLPROPERTIES (
  'write.target-file-size-bytes' = '134217728',    -- 128MB
  'write.parquet.compression-codec' = 'zstd',
  'write.parquet.compression-level' = '3',
  'write.delete.mode' = 'merge-on-read',
  'write.distribution-mode' = 'hash',
  'commit.manifest.min-count-to-merge' = '50',
  'write.metadata.delete-after-commit.enabled' = 'true',
  'write.metadata.previous-versions-max' = '20'
);

ALTER TABLE db.cdc_users SET TBLPROPERTIES (
  'write.target-file-size-bytes' = '268435456',
  'write.delete.mode' = 'merge-on-read',
  'write.update.mode' = 'merge-on-read',
  'write.merge.mode' = 'merge-on-read',
  'write.delete.target-file-size-bytes' = '67108864',
  'format-version' = '2',
  'commit.manifest-merge.enabled' = 'true'
);

-- 파일 수, 평균 크기
SELECT count(*), avg(file_size_in_bytes)
FROM iceberg.db."orders$files";
 
-- Snapshot 수
SELECT count(*) FROM iceberg.db."orders$snapshots";
 
-- 파티션별 파일 통계
SELECT partition, file_count, total_size, position_delete_record_count
FROM iceberg.db."orders$partitions"
ORDER BY file_count DESC LIMIT 20;