-- Flink SQL 批量写入示例
INSERT INTO paimon_table 
SELECT * FROM source_table;

-- 配置参数
SET 'sink.parallelism' = '16';
SET 'sink.buffer-size' = '64mb';
SET 'compaction.max-file-num' = '50';

-- 创建流式写入作业
CREATE TABLE paimon_sink (
    id BIGINT,
    name STRING,
    ts TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'paimon',
    'path' = 'hdfs://cluster/warehouse/paimon_sink',
    'write-buffer-size' = '256mb',
    'write-buffer-spillable' = 'true'
);

# 写入缓冲区配置
write-buffer-size = 256mb
write-buffer-spillable = true

# 文件大小控制
target-file-size = 128mb
compaction.max-file-num = 50

# 并发控制
sink.parallelism = 16
compaction.max-sorted-run-num = 10

# 压缩配置
file.format = parquet
parquet.compression = snappy

-- 创建 Paimon Catalog
CREATE CATALOG paimon_catalog WITH (
    'type' = 'paimon',
    'warehouse' = 'hdfs://cluster/warehouse'
);

-- 使用 Catalog
USE CATALOG paimon_catalog;

-- 查询表数据
SELECT * FROM my_database.my_table 
WHERE dt >= '2023-12-01';

// Spark SQL 配置
spark.sql.catalog.paimon = org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.paimon.warehouse = hdfs://cluster/warehouse

// 查询数据
spark.sql("SELECT * FROM paimon.db.table WHERE id > 100")

-- 查询指定快照
SELECT * FROM my_table /*+ OPTIONS('scan.snapshot-id' = '3') */;

-- 查询指定时间点
SELECT * FROM my_table FOR SYSTEM_TIME AS OF TIMESTAMP '2023-12-01 10:00:00';

-- 增量查询
SELECT * FROM my_table /*+ OPTIONS(
    'scan.mode' = 'from-snapshot',
    'scan.snapshot-id' = '1'
) */;

-- 变更查询
SELECT * FROM my_table /*+ OPTIONS('scan.mode' = 'changelog') */;

-- Flink 流式增量查询
CREATE TABLE paimon_source (
    id BIGINT,
    name STRING,
    ts TIMESTAMP(3)
) WITH (
    'connector' = 'paimon',
    'path' = 'hdfs://cluster/warehouse/my_table',
    'scan.mode' = 'latest'  -- 流式读取最新数据
);

# 查询并行度
scan.parallelism = 32

# 批次大小
scan.batch-size = 8192

# 缓存配置
lookup.cache-max-memory-size = 256mb
lookup.cache-ttl = 1h

# 统计信息检查间隔
metastore.partitioned-table.check-interval = 1d

-- 添加新列
ALTER TABLE my_table ADD COLUMN new_col STRING;

-- 删除列
ALTER TABLE my_table DROP COLUMN old_col;

-- 修改列类型
ALTER TABLE my_table ALTER COLUMN age TYPE BIGINT;

-- 重命名列
ALTER TABLE my_table RENAME COLUMN old_name TO new_name;

-- 数组类型
CREATE TABLE array_table (
    id BIGINT,
    tags ARRAY<STRING>,
    scores ARRAY<DOUBLE>
);

-- Map类型
CREATE TABLE map_table (
    id BIGINT,
    properties MAP<STRING, STRING>,
    metrics MAP<STRING, DOUBLE>
);

-- Row类型（结构体）
CREATE TABLE row_table (
    id BIGINT,
    address ROW<
        street STRING,
        city STRING,
        zipcode INT
    >
);

-- 嵌套复杂类型
CREATE TABLE nested_table (
    id BIGINT,
    user_info ROW<
        name STRING,
        contacts ARRAY<ROW<type STRING, value STRING>>,
        preferences MAP<STRING, ARRAY<STRING>>
    >
);

# Paimon 核心配置
paimon:
  # 存储配置
  warehouse: "hdfs://cluster/warehouse"
  
  # 元数据配置
  metastore:
    type: "hive"  # hive, filesystem, jdbc
    uri: "thrift://metastore:9083"
  
  # 文件系统配置
  fs:
    hdfs:
      impl: "org.apache.hadoop.hdfs.DistributedFileSystem"
    s3:
      access-key: "${S3_ACCESS_KEY}"
      secret-key: "${S3_SECRET_KEY}"
      endpoint: "s3.amazonaws.com"
  
  # 表默认配置
  table:
    default:
      file-format: "parquet"
      compression: "snappy"
      target-file-size: "128mb"
      compaction:
        max-file-num: 50
        min-file-num: 5

# Flink 作业配置
flink:
  execution:
    checkpointing:
      interval: 60s
      mode: EXACTLY_ONCE
    
  state:
    backend: rocksdb
    checkpoints:
      dir: "hdfs://cluster/checkpoints"
  
  # Paimon 连接器配置
  table:
    sql-client:
      catalogs:
        paimon:
          type: paimon
          warehouse: hdfs://cluster/warehouse

# Prometheus 监控配置
metrics:
  reporters:
    - type: prometheus
      port: 9249
      
  # 自定义指标
  custom:
    - name: paimon_write_latency
      type: histogram
      buckets: [0.1, 0.5, 1.0, 5.0, 10.0]
    
    - name: paimon_compaction_duration
      type: gauge
      
    - name: paimon_file_count
      type: counter

# 写入性能调优
write-buffer-size = 256mb
write-buffer-spillable = true
sink.parallelism = 32
compaction.max-file-num = 50

# 查询性能调优
scan.parallelism = 64
scan.batch-size = 8192
lookup.cache-max-memory-size = 512mb
lookup.cache-ttl = 2h

# 存储性能调优
target-file-size = 256mb
file.format = parquet
parquet.compression = zstd
parquet.block-size = 256mb

# JVM 性能调优
-Xmx8g -Xms8g
-XX:+UseG1GC
-XX:MaxGCPauseMillis=200
-XX:+UnlockExperimentalVMOptions
-XX:+UseZGC  # Java 11+

# 检查表状态
./paimon-admin.sh table-info --warehouse hdfs://cluster/warehouse --database mydb --table mytable

# 检查快照信息
./paimon-admin.sh snapshots --warehouse hdfs://cluster/warehouse --database mydb --table mytable

# 检查文件状态
./paimon-admin.sh files --warehouse hdfs://cluster/warehouse --database mydb --table mytable

# 修复表元数据
./paimon-admin.sh repair --warehouse hdfs://cluster/warehouse --database mydb --table mytable

# 压缩表数据
./paimon-admin.sh compact --warehouse hdfs://cluster/warehouse --database mydb --table mytable

-- 创建 MySQL CDC 源表
CREATE TABLE mysql_source (
    id BIGINT,
    name STRING,
    age INT,
    email STRING,
    update_time TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = 'password',
    'database-name' = 'mydb',
    'table-name' = 'users'
);

-- 创建 Paimon 目标表
CREATE TABLE paimon_sink (
    id BIGINT,
    name STRING,
    age INT,
    email STRING,
    update_time TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'paimon',
    'path' = 'hdfs://cluster/warehouse/mydb/users',
    'auto-create' = 'true'
);

-- 实时同步数据
INSERT INTO paimon_sink SELECT * FROM mysql_source;

# CDC 源配置
scan.startup.mode = initial
scan.incremental.snapshot.enabled = true
scan.incremental.snapshot.chunk.size = 8096

# Paimon Sink 配置
sink.parallelism = 16
sink.buffer-flush.max-rows = 10000
sink.buffer-flush.interval = 30s

# 容错配置
execution.checkpointing.interval = 60s
execution.checkpointing.mode = EXACTLY_ONCE
state.backend = rocksdb

-- ODS 层：原始数据
CREATE TABLE ods_user_behavior (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    ts TIMESTAMP(3),
    dt STRING,
    PRIMARY KEY (user_id, item_id, ts) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
    'connector' = 'paimon',
    'path' = 'hdfs://warehouse/ods/user_behavior'
);

-- DWD 层：明细数据
CREATE TABLE dwd_user_behavior (
    user_id BIGINT,
    item_id BIGINT,
    behavior_type STRING,
    behavior_time TIMESTAMP(3),
    session_id STRING,
    dt STRING,
    PRIMARY KEY (user_id, item_id, behavior_time) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
    'connector' = 'paimon',
    'path' = 'hdfs://warehouse/dwd/user_behavior'
);

-- DWS 层：汇总数据
CREATE TABLE dws_user_behavior_1h (
    user_id BIGINT,
    window_start TIMESTAMP(3),
    window_end TIMESTAMP(3),
    pv_count BIGINT,
    uv_count BIGINT,
    dt STRING,
    PRIMARY KEY (user_id, window_start) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
    'connector' = 'paimon',
    'path' = 'hdfs://warehouse/dws/user_behavior_1h'
);

-- 创建统一 Catalog
CREATE CATALOG unified_catalog WITH (
    'type' = 'hive',
    'hive-conf-dir' = '/opt/hive/conf'
);

-- Paimon 表
CREATE TABLE unified_catalog.db1.paimon_table (
    id BIGINT,
    name STRING,
    ts TIMESTAMP(3)
) WITH (
    'connector' = 'paimon',
    'path' = 'hdfs://warehouse/paimon/table1'
);

-- Iceberg 表
CREATE TABLE unified_catalog.db1.iceberg_table (
    id BIGINT,
    name STRING,
    ts TIMESTAMP(3)
) WITH (
    'connector' = 'iceberg',
    'catalog-name' = 'hive_catalog',
    'warehouse' = 'hdfs://warehouse/iceberg'
);

-- 跨表查询
SELECT p.id, p.name, i.name as iceberg_name
FROM unified_catalog.db1.paimon_table p
JOIN unified_catalog.db1.iceberg_table i
ON p.id = i.id;

# 生产环境 Paimon 配置
paimon:
  # 基础配置
  warehouse: "hdfs://prod-cluster/warehouse"
  
  # 性能配置
  table:
    default:
      # 文件配置
      target-file-size: "256mb"
      file.format: "parquet"
      parquet.compression: "zstd"
      
      # 压缩配置
      compaction:
        max-file-num: 30
        min-file-num: 5
        max-size-amplification-percent: 200
        
      # 写入配置
      write-buffer-size: "512mb"
      write-buffer-spillable: true
      
      # 查询配置
      scan.parallelism: 64
      lookup.cache-max-memory-size: "1gb"
      
  # 清理配置
  snapshot:
    time-retained: "7d"
    num-retained-max: 100
    
  # 监控配置
  metrics:
    reporter: "prometheus"
    port: 9249

# 增大写入缓冲区
write-buffer-size = 512mb
write-buffer-spillable = true

# 调整并行度
sink.parallelism = 32

# 优化批次大小
sink.buffer-flush.max-rows = 20000
sink.buffer-flush.interval = 30s

-- 分区剪枝
SELECT * FROM table WHERE dt BETWEEN '2023-12-01' AND '2023-12-31';

-- 投影下推
SELECT id, name FROM table WHERE status = 'active';

-- 谓词下推
SELECT * FROM table WHERE age > 18 AND city = 'Beijing';

# 查询缓存配置
lookup.cache-max-memory-size = 1gb
lookup.cache-ttl = 2h

# 元数据缓存
metastore.cache.expiration-interval-ms = 600000

-- 开启 MySQL binlog
SET GLOBAL binlog_format = 'ROW';
SET GLOBAL binlog_row_image = 'FULL';

-- MySQL CDC 源表
CREATE TABLE mysql_users (
    id BIGINT,
    name STRING,
    email STRING,
    created_at TIMESTAMP(3),
    updated_at TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'cdc_user',
    'password' = 'password',
    'database-name' = 'mydb',
    'table-name' = 'users',
    'scan.startup.mode' = 'initial'
);

-- Paimon 目标表
CREATE TABLE paimon_users (
    id BIGINT,
    name STRING,
    email STRING,
    created_at TIMESTAMP(3),
    updated_at TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'paimon',
    'path' = 'hdfs://cluster/warehouse/mydb/users'
);

-- 同步作业
INSERT INTO paimon_users SELECT * FROM mysql_users;

-- 添加新列（向后兼容）
ALTER TABLE my_table ADD COLUMN new_field STRING;

-- 删除列（向前兼容）
ALTER TABLE my_table DROP COLUMN old_field;

-- 扩大类型（兼容）
ALTER TABLE my_table ALTER COLUMN age TYPE BIGINT;

-- 缩小类型（需要验证）
-- 先检查数据范围，再执行变更

# 关键监控指标
metrics:
  # 性能指标
  - write_latency_p99
  - query_latency_p95
  - throughput_per_second
  
  # 资源指标
  - cpu_usage_percent
  - memory_usage_percent
  - disk_io_utilization
  
  # 业务指标
  - file_count_per_table
  - compaction_frequency
  - snapshot_count

# 表状态检查
./paimon-admin.sh table-info --table my_table

# 手动触发压缩
./paimon-admin.sh compact --table my_table

# 快照清理
./paimon-admin.sh expire-snapshots --table my_table --retain-max 50