// Flink Exactly-Once 配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 1. 启用 Checkpoint
env.enableCheckpointing(60000);  // 60 秒
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 2. Kafka Source Exactly-Once
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
    "topic",
    new SimpleStringSchema(),
    properties
);
consumer.setStartFromGroupOffsets();  // 从 Group Offset 开始

// 3. Kafka Sink Exactly-Once
KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("localhost:9092")
    .setRecordSerializer(...)
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)  // 精确一次
    .setTransactionalIdPrefix("flink-kafka-")
    .build();

// 使用唯一 ID 保证幂等性
public class IdempotentSinkFunction extends RichSinkFunction<Order> {
    private transient RedisClient redis;
    
    @Override
    public void invoke(Order order, Context context) {
        String key = "order:" + order.getOrderId();
        
        // 检查是否已处理
        if (redis.exists(key)) {
            return;  // 已处理，跳过
        }
        
        // 处理数据
        processOrder(order);
        
        // 标记已处理
        redis.setex(key, 3600, "processed");  // 1 小时过期
    }
}

public class LatencyMonitor extends ProcessFunction<Event, Event> {
    private transient Histogram latencyHistogram;
    
    @Override
    public void open(Configuration parameters) {
        latencyHistogram = getRuntimeContext()
            .getMetricGroup()
            .histogram("end_to_end_latency", 
                new DescriptiveStatisticsHistogram(1000));
    }
    
    @Override
    public void processElement(Event event, Context ctx, Collector<Event> out) {
        long latency = System.currentTimeMillis() - event.getTimestamp();
        latencyHistogram.update(latency);
        
        if (latency > 5000) {  // 超过 5 秒告警
            log.warn("High latency detected: {}ms, event: {}", latency, event);
        }
        
        out.collect(event);
    }
}

// 设置 Watermark 策略
WatermarkStrategy<Order> watermarkStrategy = WatermarkStrategy
    .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(10))  // 容忍 10 秒乱序
    .withTimestampAssigner((order, timestamp) -> order.getCreateTime())
    .withIdleness(Duration.ofMinutes(1));  // 1 分钟无数据则推进 Watermark

DataStream<Order> watermarkedStream = stream
    .assignTimestampsAndWatermarks(watermarkStrategy);

// 窗口允许延迟
DataStream<OrderStats> result = watermarkedStream
    .keyBy(Order::getUserId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .allowedLateness(Time.minutes(1))  // 允许 1 分钟延迟
    .sideOutputLateData(lateOutputTag)  // 超过延迟的数据输出到侧输出流
    .aggregate(new OrderAggregateFunction());

// 处理迟到数据
DataStream<Order> lateData = result.getSideOutput(lateOutputTag);
lateData.addSink(new LateDataSink());  // 单独处理迟到数据

// RocksDB 状态后端配置
EmbeddedRocksDBStateBackend backend = new EmbeddedRocksDBStateBackend(true);  // 启用增量 Checkpoint
backend.setDbStoragePath("/data/flink/rocksdb");

env.setStateBackend(backend);
env.getCheckpointConfig().setCheckpointStorage("hdfs://namenode:9000/flink/checkpoints");

// 设置状态 TTL
StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.hours(24))  // 24 小时过期
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)  // 创建和写入时更新
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)  // 不返回过期数据
    .cleanupFullSnapshot()  // 全量快照时清理
    .cleanupIncrementally(10, true)  // 增量清理
    .build();

ValueStateDescriptor<String> descriptor = 
    new ValueStateDescriptor<>("user-state", String.class);
descriptor.enableTimeToLive(ttlConfig);

ValueState<String> state = getRuntimeContext().getState(descriptor);

// 批处理层：Spark Batch 计算每日订单统计
SparkSession spark = SparkSession.builder()
    .appName("Daily Order Stats")
    .getOrCreate();

Dataset<Row> orders = spark.read()
    .format("hive")
    .table("ods.orders")
    .where("dt = '2024-02-28'");

Dataset<Row> stats = orders
    .groupBy("province", "category")
    .agg(
        count("order_id").as("order_count"),
        sum("amount").as("total_amount")
    );

stats.write()
    .mode(SaveMode.Overwrite)
    .format("hive")
    .saveAsTable("ads.daily_order_stats");

// 速度层：Flink 实时计算订单统计
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<Order> orders = env
    .addSource(new FlinkKafkaConsumer<>("orders", ...));

DataStream<OrderStats> realtimeStats = orders
    .keyBy(order -> Tuple2.of(order.getProvince(), order.getCategory()))
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new OrderAggregateFunction());

realtimeStats.addSink(new RedisSink<>(...));

// 服务层：合并批处理和实时视图
@GetMapping("/order-stats")
public OrderStats getOrderStats(String province, String category) {
    // 1. 从 Hive 获取批处理结果（昨天及之前）
    OrderStats batchStats = hiveService.getStats(province, category, 
        LocalDate.now().minusDays(1));
    
    // 2. 从 Redis 获取实时结果（今天）
    OrderStats realtimeStats = redisService.getStats(province, category);
    
    // 3. 合并结果
    return OrderStats.builder()
        .orderCount(batchStats.getOrderCount() + realtimeStats.getOrderCount())
        .totalAmount(batchStats.getTotalAmount().add(realtimeStats.getTotalAmount()))
        .build();
}

// 统一的流处理代码
public class OrderStatsJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // Kafka 配置
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "order-stats");
        
        // 根据参数决定是实时处理还是历史回溯
        FlinkKafkaConsumer<Order> consumer = new FlinkKafkaConsumer<>(
            "orders",
            new OrderDeserializationSchema(),
            properties
        );
        
        if (args.length > 0 && "reprocess".equals(args[0])) {
            // 历史回溯：从指定 offset 开始
            Map<TopicPartition, Long> specificStartOffsets = new HashMap<>();
            specificStartOffsets.put(new TopicPartition("orders", 0), 0L);
            specificStartOffsets.put(new TopicPartition("orders", 1), 0L);
            consumer.setStartFromSpecificOffsets(specificStartOffsets);
        } else {
            // 实时处理：从最新 offset 开始
            consumer.setStartFromLatest();
        }
        
        // 统一的处理逻辑
        DataStream<Order> orders = env.addSource(consumer);
        
        DataStream<OrderStats> stats = orders
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                    .withTimestampAssigner((order, ts) -> order.getCreateTime())
            )
            .keyBy(order -> Tuple2.of(order.getProvince(), order.getCategory()))
            .window(TumblingEventTimeWindows.of(Time.hours(1)))
            .aggregate(new OrderAggregateFunction());
        
        // 写入存储
        stats.addSink(new ClickHouseSink<>(...));
        
        env.execute("Order Stats Job");
    }
}

// Kafka 配置：长期保留
// server.properties
log.retention.hours=720  # 保留 30 天
log.segment.bytes=1073741824  # 1GB
log.retention.check.interval.ms=300000

// Flink 流批一体示例
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 流处理：无界流
DataStream<Order> streamOrders = env
    .addSource(new FlinkKafkaConsumer<>("orders", ...));

// 批处理：有界流
DataStream<Order> batchOrders = env
    .readTextFile("hdfs://path/to/orders")
    .map(new OrderMapper());

// 统一的处理逻辑
DataStream<OrderStats> processOrders(DataStream<Order> orders) {
    return orders
        .keyBy(Order::getUserId)
        .window(TumblingEventTimeWindows.of(Time.hours(1)))
        .aggregate(new OrderAggregateFunction());
}

DataStream<OrderStats> streamStats = processOrders(streamOrders);
DataStream<OrderStats> batchStats = processOrders(batchOrders);

# canal.properties
canal.id = 1
canal.ip = 192.168.1.100
canal.port = 11111
canal.zkServers = localhost:2181

# instance.properties
canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset=UTF-8

# 订阅的数据库和表
canal.instance.filter.regex=test_db\\..*
canal.instance.filter.black.regex=test_db\\.tmp_.*

# Kafka 配置
canal.mq.servers=localhost:9092
canal.mq.topic=canal_topic
canal.mq.partition=0

{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "localhost",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",
    "database.server.name": "mysql-server",
    "database.include.list": "inventory",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.inventory",
    "include.schema.changes": "true",
    "transforms": "route",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
    "transforms.route.replacement": "$3"
  }
}

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;

public class FlinkCDCExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 配置 MySQL CDC Source
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("localhost")
            .port(3306)
            .databaseList("inventory")  // 监控的数据库
            .tableList("inventory.orders", "inventory.products")  // 监控的表
            .username("root")
            .password("password")
            .startupOptions(StartupOptions.initial())  // 全量 + 增量
            .deserializer(new JsonDebeziumDeserializationSchema())  // JSON 格式
            .build();
        
        // 创建数据流
        DataStreamSource<String> cdcStream = env
            .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC");
        
        // 处理 CDC 数据
        cdcStream
            .map(new ParseCDCFunction())
            .keyBy(Order::getOrderId)
            .process(new OrderProcessFunction())
            .addSink(new KafkaSink<>(...));
        
        env.execute("Flink CDC Job");
    }
}

// 解析 CDC 事件
public class ParseCDCFunction implements MapFunction<String, Order> {
    @Override
    public Order map(String value) throws Exception {
        JSONObject json = JSON.parseObject(value);
        String op = json.getString("op");  // c/u/d/r
        
        switch (op) {
            case "c":  // CREATE (INSERT)
            case "r":  // READ (全量同步)
                return parseOrder(json.getJSONObject("after"));
            case "u":  // UPDATE
                return parseOrder(json.getJSONObject("after"));
            case "d":  // DELETE
                Order order = parseOrder(json.getJSONObject("before"));
                order.setDeleted(true);
                return order;
            default:
                return null;
        }
    }
}

# Broker 配置
# 网络线程数
num.network.threads=8
# IO 线程数
num.io.threads=16
# Socket 缓冲区
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

# 日志配置
# 保留时间（7 天）
log.retention.hours=168
# 日志段大小（1GB）
log.segment.bytes=1073741824
# 检查间隔
log.retention.check.interval.ms=300000

# 副本配置
# 默认副本数
default.replication.factor=3
# 最小同步副本数
min.insync.replicas=2
# 副本延迟阈值
replica.lag.time.max.ms=10000

# 性能优化
# 压缩类型
compression.type=lz4
# 批量大小
batch.size=16384
# 等待时间
linger.ms=10
# 缓冲区大小
buffer.memory=33554432

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 可靠性配置
props.put("acks", "all");  // 所有副本确认
props.put("retries", 3);  // 重试次数
props.put("max.in.flight.requests.per.connection", 1);  // 保证顺序

// 性能配置
props.put("compression.type", "lz4");  // 压缩
props.put("batch.size", 16384);  // 批量大小
props.put("linger.ms", 10);  // 等待时间
props.put("buffer.memory", 33554432);  // 缓冲区

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// 异步发送
producer.send(new ProducerRecord<>("topic", key, value), (metadata, exception) -> {
    if (exception != null) {
        log.error("Send failed", exception);
    } else {
        log.info("Sent to partition {} offset {}", 
            metadata.partition(), metadata.offset());
    }
});

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "order-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

// 消费配置
props.put("enable.auto.commit", false);  // 手动提交
props.put("auto.offset.reset", "earliest");  // 从最早开始
props.put("max.poll.records", 500);  // 单次拉取记录数
props.put("fetch.min.bytes", 1024);  // 最小拉取字节数
props.put("fetch.max.wait.ms", 500);  // 最大等待时间

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("orders"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 处理消息
        processRecord(record);
    }
    // 手动提交
    consumer.commitSync();
}

// 自定义分区器
public class CustomPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, 
                        Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        
        // 根据业务逻辑自定义分区
        if (key == null) {
            return ThreadLocalRandom.current().nextInt(numPartitions);
        }
        
        // 例如：VIP 用户到特定分区
        if (isVIP(key.toString())) {
            return 0;  // VIP 专属分区
        }
        
        // 普通用户哈希分区
        return Math.abs(key.hashCode()) % (numPartitions - 1) + 1;
    }
}

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Source
DataStream<String> text = env.readTextFile("input.txt");

// Transformation
DataStream<Tuple2<String, Integer>> counts = text
    .flatMap(new Tokenizer())
    .keyBy(value -> value.f0)
    .sum(1);

// Sink
counts.print();

env.execute("WordCount");

// 滚动窗口：每 5 分钟统计一次
DataStream<OrderStats> tumblingStats = orders
    .keyBy(Order::getUserId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new OrderAggregateFunction());

// 滑动窗口：最近 10 分钟，每 5 分钟更新一次
DataStream<OrderStats> slidingStats = orders
    .keyBy(Order::getUserId)
    .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(5)))
    .aggregate(new OrderAggregateFunction());

// 会话窗口：30 分钟无活动则结束会话
DataStream<SessionStats> sessionStats = behaviors
    .keyBy(Behavior::getUserId)
    .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
    .aggregate(new SessionAggregateFunction());

// 计数窗口：每 100 条数据触发一次
DataStream<OrderStats> countStats = orders
    .keyBy(Order::getUserId)
    .countWindow(100)
    .aggregate(new OrderAggregateFunction());

// 设置 Watermark 策略
WatermarkStrategy<Order> watermarkStrategy = WatermarkStrategy
    .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(10))  // 容忍 10 秒乱序
    .withTimestampAssigner((order, timestamp) -> order.getCreateTime())  // 提取时间戳
    .withIdleness(Duration.ofMinutes(1));  // 1 分钟无数据则推进 Watermark

DataStream<Order> watermarkedStream = stream
    .assignTimestampsAndWatermarks(watermarkStrategy);

// ValueState：单值状态
ValueStateDescriptor<Long> descriptor = 
    new ValueStateDescriptor<>("count", Long.class);
ValueState<Long> countState = getRuntimeContext().getState(descriptor);

Long count = countState.value();
countState.update(count == null ? 1 : count + 1);

// ListState：列表状态
ListStateDescriptor<String> listDescriptor = 
    new ListStateDescriptor<>("list", String.class);
ListState<String> listState = getRuntimeContext().getListState(listDescriptor);

listState.add("item");
for (String item : listState.get()) {
    // 处理列表项
}

// MapState：映射状态
MapStateDescriptor<String, Long> mapDescriptor = 
    new MapStateDescriptor<>("map", String.class, Long.class);
MapState<String, Long> mapState = getRuntimeContext().getMapState(mapDescriptor);

mapState.put("key", 100L);
Long value = mapState.get("key");

// ReducingState：聚合状态
ReducingStateDescriptor<Long> reducingDescriptor = 
    new ReducingStateDescriptor<>("sum", (a, b) -> a + b, Long.class);
ReducingState<Long> reducingState = getRuntimeContext().getReducingState(reducingDescriptor);

reducingState.add(10L);
Long sum = reducingState.get();

// Checkpoint 配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 启用 Checkpoint
env.enableCheckpointing(60000);  // 60 秒

// Checkpoint 模式
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// Checkpoint 超时
env.getCheckpointConfig().setCheckpointTimeout(600000);  // 10 分钟

// 最小间隔
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);  // 30 秒

// 最大并发 Checkpoint
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// 容忍失败次数
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);

// 外部化 Checkpoint
env.getCheckpointConfig().enableExternalizedCheckpoints(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

// Checkpoint 存储
env.getCheckpointConfig().setCheckpointStorage("hdfs://namenode:9000/flink/checkpoints");

-- MergeTree：订单明细表
CREATE TABLE orders (
    order_id String,
    user_id String,
    product_id String,
    amount Decimal(18, 2),
    status String,
    province String,
    city String,
    create_time DateTime,
    update_time DateTime
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(create_time)  -- 按月分区
ORDER BY (user_id, order_id, create_time)  -- 排序键
SETTINGS index_granularity = 8192;  -- 索引粒度

-- ReplacingMergeTree：用户信息表（去重）
CREATE TABLE user_profile (
    user_id String,
    name String,
    age UInt8,
    gender String,
    province String,
    city String,
    update_time DateTime
) ENGINE = ReplacingMergeTree(update_time)  -- 按 update_time 去重
ORDER BY user_id;

-- SummingMergeTree：订单统计表（自动聚合）
CREATE TABLE order_stats (
    date Date,
    province String,
    category String,
    order_count UInt64,
    total_amount Decimal(18, 2)
) ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, province, category);

-- AggregatingMergeTree：用户行为统计（复杂聚合）
CREATE TABLE user_behavior_stats (
    date Date,
    user_id String,
    pv SimpleAggregateFunction(sum, UInt64),  -- 简单聚合
    uv AggregateFunction(uniq, String)  -- 复杂聚合（去重计数）
) ENGINE = AggregatingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, user_id);

-- 按月分区（推荐）
PARTITION BY toYYYYMM(create_time)

-- 按天分区（数据量大时）
PARTITION BY toYYYYMMDD(create_time)

-- 按小时分区（实时场景）
PARTITION BY toYYYYMMDDhh(create_time)

-- 多级分区
PARTITION BY (toYYYYMM(create_time), province)

-- 查看分区
SELECT 
    partition,
    name,
    rows,
    bytes_on_disk,
    formatReadableSize(bytes_on_disk) AS size
FROM system.parts
WHERE table = 'orders' AND active
ORDER BY partition DESC;

-- 删除旧分区
ALTER TABLE orders DROP PARTITION '202401';

-- 优化分区（合并小文件）
OPTIMIZE TABLE orders PARTITION '202402';

-- 主键索引（ORDER BY）
ORDER BY (user_id, create_time)

-- 跳数索引（Skip Index）
-- minmax：数值范围查询
ALTER TABLE orders ADD INDEX idx_amount amount TYPE minmax GRANULARITY 4;

-- set：低基数字段
ALTER TABLE orders ADD INDEX idx_province province TYPE set(100) GRANULARITY 4;

-- bloom_filter：等值查询
ALTER TABLE orders ADD INDEX idx_status status TYPE bloom_filter GRANULARITY 4;

-- ngrambf_v1：模糊查询
ALTER TABLE orders ADD INDEX idx_user_name user_name TYPE ngrambf_v1(3, 256, 2, 0) GRANULARITY 4;

-- 物化列（Materialized Column）
ALTER TABLE orders ADD COLUMN order_date Date MATERIALIZED toDate(create_time);

-- 物化视图（Materialized View）
CREATE MATERIALIZED VIEW order_stats_mv
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, province)
AS SELECT
    toDate(create_time) AS date,
    province,
    count() AS order_count,
    sum(amount) AS total_amount
FROM orders
GROUP BY date, province;

-- 使用 PREWHERE 过滤（比 WHERE 快）
SELECT *
FROM orders
PREWHERE create_time >= '2024-01-01'  -- 先过滤
WHERE amount > 100;  -- 再过滤

-- 避免 SELECT *
SELECT order_id, amount, create_time
FROM orders;

-- 使用 FINAL 去重（ReplacingMergeTree）
SELECT *
FROM user_profile
FINAL
WHERE user_id = 'user123';

-- 分布式查询优化
SELECT province, sum(amount)
FROM orders
WHERE create_time >= today()
GROUP BY province
SETTINGS distributed_aggregation_memory_efficient = 1;

-- 使用物化视图
SELECT date, province, order_count, total_amount
FROM order_stats_mv
WHERE date >= '2024-01-01';

-- 并行查询
SELECT *
FROM orders
SETTINGS max_threads = 8;

-- 批量写入（推荐 10000-100000 行）
INSERT INTO orders VALUES
    ('order1', 'user1', 'product1', 100.00, 'PAID', '北京', '北京', '2024-01-01 10:00:00', '2024-01-01 10:00:00'),
    ('order2', 'user2', 'product2', 200.00, 'PAID', '上海', '上海', '2024-01-01 10:01:00', '2024-01-01 10:01:00'),
    -- ... 更多数据
    ;

-- 异步写入
INSERT INTO orders SETTINGS async_insert = 1, wait_for_async_insert = 0
VALUES (...);

-- 并行写入（多个客户端同时写入不同分区）

-- 写入性能监控
SELECT
    table,
    sum(rows) AS total_rows,
    sum(bytes) AS total_bytes,
    formatReadableSize(total_bytes) AS size,
    count() AS parts_count
FROM system.parts
WHERE table = 'orders' AND active
GROUP BY table;

// Flink 写入 Hudi
import org.apache.hudi.configuration.FlinkOptions

val hoodieOptions = Map(
  FlinkOptions.PATH.key() -> "hdfs://namenode:9000/hudi/orders",
  FlinkOptions.TABLE_NAME.key() -> "orders",
  FlinkOptions.TABLE_TYPE.key() -> "MERGE_ON_READ",  // 表类型
  FlinkOptions.PRECOMBINE_FIELD.key() -> "update_time",  // 去重字段
  FlinkOptions.RECORD_KEY_FIELD.key() -> "order_id",  // 主键
  FlinkOptions.PARTITION_PATH_FIELD.key() -> "dt",  // 分区字段
  FlinkOptions.OPERATION.key() -> "upsert",  // 操作类型
  FlinkOptions.WRITE_TASKS.key() -> "4",  // 写入并行度
  FlinkOptions.COMPACTION_TASKS.key() -> "2"  // 压缩并行度
)

dataStream
  .addSink(
    HoodieFlinkStreamer.builder()
      .withOptions(hoodieOptions)
      .build()
  )

// Spark 读取 Hudi
val df = spark.read
  .format("hudi")
  .load("hdfs://namenode:9000/hudi/orders")

// 增量查询
val incrementalDF = spark.read
  .format("hudi")
  .option("hoodie.datasource.query.type", "incremental")
  .option("hoodie.datasource.read.begin.instanttime", "20240101000000")
  .load("hdfs://namenode:9000/hudi/orders")

// 时间旅行
val historicalDF = spark.read
  .format("hudi")
  .option("as.of.instant", "20240101000000")
  .load("hdfs://namenode:9000/hudi/orders")

-- 创建表时定义分区
CREATE TABLE orders (
    order_id STRING,
    user_id STRING,
    amount DECIMAL(18, 2),
    create_time TIMESTAMP
) USING iceberg
PARTITIONED BY (days(create_time));  -- 按天分区

-- 查询时无需指定分区
SELECT * FROM orders WHERE create_time >= '2024-01-01';

-- 添加列
ALTER TABLE orders ADD COLUMN province STRING;

-- 删除列
ALTER TABLE orders DROP COLUMN province;

-- 重命名列
ALTER TABLE orders RENAME COLUMN amount TO order_amount;

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

-- 查询指定时间点的数据
SELECT * FROM orders FOR SYSTEM_TIME AS OF '2024-01-01 00:00:00';

-- 查询指定快照的数据
SELECT * FROM orders FOR SYSTEM_VERSION AS OF 123456789;

-- 查看快照历史
SELECT * FROM orders.snapshots;

-- Kafka Topic: ods_order_topic
-- ClickHouse 表结构
CREATE TABLE ods_order (
    -- 业务字段
    order_id String,
    user_id String,
    product_id String,
    amount Decimal(18, 2),
    status String,
    province String,
    city String,
    create_time DateTime,
    update_time DateTime,
    
    -- 元数据字段
    source_db String,              -- 来源数据库
    source_table String,           -- 来源表
    binlog_file String,            -- Binlog 文件
    binlog_pos UInt64,             -- Binlog 位置
    event_type String,             -- 事件类型 (INSERT/UPDATE/DELETE)
    event_time DateTime,           -- 事件时间
    
    -- 系统字段
    etl_time DateTime DEFAULT now(),  -- 数据采集时间
    etl_date Date DEFAULT today()     -- 分区字段
) ENGINE = MergeTree()
PARTITION BY etl_date
ORDER BY (order_id, create_time);

public class OdsWriterJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // Flink CDC 读取 MySQL
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("localhost")
            .port(3306)
            .databaseList("ecommerce")
            .tableList("ecommerce.orders")
            .username("root")
            .password("password")
            .deserializer(new JsonDebeziumDeserializationSchema())
            .build();
        
        DataStream<String> cdcStream = env
            .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC");
        
        // 解析 CDC 事件，添加元数据
        DataStream<OdsOrder> odsStream = cdcStream
            .map(new RichMapFunction<String, OdsOrder>() {
                @Override
                public OdsOrder map(String value) throws Exception {
                    JSONObject json = JSON.parseObject(value);
                    JSONObject source = json.getJSONObject("source");
                    JSONObject after = json.getJSONObject("after");
                    
                    OdsOrder order = new OdsOrder();
                    // 业务字段
                    order.setOrderId(after.getString("order_id"));
                    order.setUserId(after.getString("user_id"));
                    order.setAmount(after.getBigDecimal("amount"));
                    // ... 其他字段
                    
                    // 元数据字段
                    order.setSourceDb(source.getString("db"));
                    order.setSourceTable(source.getString("table"));
                    order.setBinlogFile(source.getString("file"));
                    order.setBinlogPos(source.getLong("pos"));
                    order.setEventType(json.getString("op"));
                    order.setEventTime(source.getLong("ts_ms"));
                    order.setEtlTime(System.currentTimeMillis());
                    
                    return order;
                }
            });
        
        // 写入 Kafka ODS Topic
        odsStream.sinkTo(
            KafkaSink.<OdsOrder>builder()
                .setBootstrapServers("localhost:9092")
                .setRecordSerializer(
                    KafkaRecordSerializationSchema.builder()
                        .setTopic("ods_order_topic")
                        .setValueSerializationSchema(new JsonSerializationSchema<>())
                        .build()
                )
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                .build()
        );
        
        env.execute("ODS Writer Job");
    }
}

-- ClickHouse DWD 层表
CREATE TABLE dwd_order (
    -- 订单信息
    order_id String,
    order_no String,
    order_type String,
    
    -- 用户信息（关联维度）
    user_id String,
    user_name String,
    user_level String,
    user_type String,
    
    -- 商品信息（关联维度）
    product_id String,
    product_name String,
    category_id String,
    category_name String,
    brand_id String,
    brand_name String,
    
    -- 金额信息
    original_amount Decimal(18, 2),
    discount_amount Decimal(18, 2),
    final_amount Decimal(18, 2),
    
    -- 状态信息
    status String,
    pay_type String,
    
    -- 地域信息
    province String,
    city String,
    district String,
    
    -- 时间信息
    create_time DateTime,
    pay_time DateTime,
    update_time DateTime,
    
    -- 系统字段
    etl_time DateTime DEFAULT now(),
    etl_date Date DEFAULT today()
) ENGINE = MergeTree()
PARTITION BY etl_date
ORDER BY (user_id, order_id, create_time);

public class DwdProcessJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 读取 ODS 层数据
        DataStream<OdsOrder> odsStream = env
            .addSource(new FlinkKafkaConsumer<>("ods_order_topic", ...));
        
        // 读取用户维度表（广播流）
        DataStream<DimUser> userStream = env
            .addSource(new FlinkKafkaConsumer<>("dim_user_topic", ...));
        
        MapStateDescriptor<String, DimUser> userStateDescriptor = 
            new MapStateDescriptor<>("user-broadcast-state", String.class, DimUser.class);
        
        BroadcastStream<DimUser> userBroadcast = userStream.broadcast(userStateDescriptor);
        
        // 关联维度表
        DataStream<DwdOrder> dwdStream = odsStream
            .connect(userBroadcast)
            .process(new BroadcastProcessFunction<OdsOrder, DimUser, DwdOrder>() {
                
                @Override
                public void processElement(OdsOrder order, ReadOnlyContext ctx, 
                        Collector<DwdOrder> out) throws Exception {
                    // 从广播状态中获取用户信息
                    ReadOnlyBroadcastState<String, DimUser> state = 
                        ctx.getBroadcastState(userStateDescriptor);
                    DimUser user = state.get(order.getUserId());
                    
                    if (user != null) {
                        DwdOrder dwdOrder = new DwdOrder();
                        // 订单信息
                        dwdOrder.setOrderId(order.getOrderId());
                        dwdOrder.setAmount(order.getAmount());
                        
                        // 用户信息
                        dwdOrder.setUserId(user.getUserId());
                        dwdOrder.setUserName(user.getUserName());
                        dwdOrder.setUserLevel(user.getUserLevel());
                        
                        out.collect(dwdOrder);
                    }
                }
                
                @Override
                public void processBroadcastElement(DimUser user, Context ctx, 
                        Collector<DwdOrder> out) throws Exception {
                    // 更新广播状态
                    ctx.getBroadcastState(userStateDescriptor).put(user.getUserId(), user);
                }
            })
            // 数据清洗
            .filter(order -> order.getAmount() != null && order.getAmount().compareTo(BigDecimal.ZERO) > 0)
            .filter(order -> order.getUserId() != null)
            // 数据转换
            .map(new OrderTransformMapper());
        
        // 写入 ClickHouse DWD 层
        dwdStream.addSink(new ClickHouseSink<>(...));
        
        env.execute("DWD Process Job");
    }
}

-- 订单分钟级汇总表
CREATE TABLE dws_order_1min (
    window_start DateTime,
    window_end DateTime,
    province String,
    city String,
    category String,
    
    -- 聚合指标
    order_count UInt64,
    user_count UInt64,
    product_count UInt64,
    total_amount Decimal(18, 2),
    avg_amount Decimal(18, 2),
    max_amount Decimal(18, 2),
    min_amount Decimal(18, 2),
    
    etl_time DateTime DEFAULT now()
) ENGINE = SummingMergeTree()
PARTITION BY toYYYYMMDD(window_start)
ORDER BY (window_start, province, city, category);

-- 用户小时级汇总表
CREATE TABLE dws_user_1hour (
    window_start DateTime,
    window_end DateTime,
    user_id String,
    user_level String,
    
    -- 行为指标
    pv UInt64,
    uv UInt64,
    click_count UInt64,
    cart_count UInt64,
    order_count UInt64,
    
    -- 金额指标
    total_amount Decimal(18, 2),
    avg_order_amount Decimal(18, 2),
    
    -- 转化指标
    click_rate Float64,
    cart_rate Float64,
    conversion_rate Float64,
    
    etl_time DateTime DEFAULT now()
) ENGINE = ReplacingMergeTree(etl_time)
PARTITION BY toYYYYMMDD(window_start)
ORDER BY (window_start, user_id);

public class DwsAggregateJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 读取 DWD 层数据
        DataStream<DwdOrder> dwdStream = env
            .addSource(new FlinkKafkaConsumer<>("dwd_order_topic", ...));
        
        // 设置 Watermark
        DataStream<DwdOrder> watermarkedStream = dwdStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<DwdOrder>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                    .withTimestampAssigner((order, ts) -> order.getCreateTime())
            );
        
        // 按维度分组 + 滚动窗口聚合
        DataStream<DwsOrderStats> dwsStream = watermarkedStream
            .keyBy(order -> Tuple3.of(
                order.getProvince(),
                order.getCity(),
                order.getCategory()
            ))
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .aggregate(new AggregateFunction<DwdOrder, OrderAccumulator, DwsOrderStats>() {
                
                @Override
                public OrderAccumulator createAccumulator() {
                    return new OrderAccumulator();
                }
                
                @Override
                public OrderAccumulator add(DwdOrder order, OrderAccumulator acc) {
                    acc.orderCount++;
                    acc.userSet.add(order.getUserId());
                    acc.productSet.add(order.getProductId());
                    acc.totalAmount = acc.totalAmount.add(order.getFinalAmount());
                    acc.maxAmount = acc.maxAmount.max(order.getFinalAmount());
                    acc.minAmount = acc.minAmount.min(order.getFinalAmount());
                    return acc;
                }
                
                @Override
                public DwsOrderStats getResult(OrderAccumulator acc) {
                    return DwsOrderStats.builder()
                        .orderCount(acc.orderCount)
                        .userCount(acc.userSet.size())
                        .productCount(acc.productSet.size())
                        .totalAmount(acc.totalAmount)
                        .avgAmount(acc.totalAmount.divide(
                            BigDecimal.valueOf(acc.orderCount), 2, RoundingMode.HALF_UP))
                        .maxAmount(acc.maxAmount)
                        .minAmount(acc.minAmount)
                        .build();
                }
                
                @Override
                public OrderAccumulator merge(OrderAccumulator a, OrderAccumulator b) {
                    a.orderCount += b.orderCount;
                    a.userSet.addAll(b.userSet);
                    a.productSet.addAll(b.productSet);
                    a.totalAmount = a.totalAmount.add(b.totalAmount);
                    a.maxAmount = a.maxAmount.max(b.maxAmount);
                    a.minAmount = a.minAmount.min(b.minAmount);
                    return a;
                }
            });
        
        // 写入 ClickHouse DWS 层
        dwsStream.addSink(new ClickHouseSink<>(...));
        
        env.execute("DWS Aggregate Job");
    }
}

-- 实时 GMV 大屏表
CREATE TABLE ads_realtime_gmv (
    stat_time DateTime,
    time_granularity String,  -- 时间粒度 (1min/5min/1hour)
    
    -- 核心指标
    gmv Decimal(18, 2),
    order_count UInt64,
    user_count UInt64,
    product_count UInt64,
    avg_order_amount Decimal(18, 2),
    
    -- 同比环比
    gmv_yoy Decimal(10, 2),  -- GMV 同比增长率
    gmv_mom Decimal(10, 2),  -- GMV 环比增长率
    order_yoy Decimal(10, 2),
    order_mom Decimal(10, 2),
    
    -- TOP 数据
    top_province String,
    top_category String,
    top_product String,
    
    -- 系统字段
    update_time DateTime DEFAULT now()
) ENGINE = ReplacingMergeTree(update_time)
ORDER BY (stat_time, time_granularity);

-- 用户画像表
CREATE TABLE ads_user_portrait (
    user_id String,
    
    -- 基本信息
    user_name String,
    age UInt8,
    gender String,
    province String,
    city String,
    
    -- 行为特征
    total_pv UInt64,
    total_uv UInt64,
    avg_session_duration UInt32,
    last_visit_time DateTime,
    
    -- 消费特征
    total_order_count UInt64,
    total_amount Decimal(18, 2),
    avg_order_amount Decimal(18, 2),
    max_order_amount Decimal(18, 2),
    last_order_time DateTime,
    
    -- 偏好特征
    favorite_category Array(String),
    favorite_brand Array(String),
    
    -- 标签
    user_tags Array(String),
    user_level String,
    
    -- 系统字段
    update_time DateTime DEFAULT now()
) ENGINE = ReplacingMergeTree(update_time)
ORDER BY user_id;

// 数据质量检查
public class DataQualityChecker extends ProcessFunction<Order, Order> {
    
    private transient Counter validCounter;
    private transient Counter invalidCounter;
    private transient Histogram qualityScore;
    
    @Override
    public void open(Configuration parameters) {
        MetricGroup metricGroup = getRuntimeContext().getMetricGroup();
        validCounter = metricGroup.counter("valid_records");
        invalidCounter = metricGroup.counter("invalid_records");
        qualityScore = metricGroup.histogram("quality_score", 
            new DescriptiveStatisticsHistogram(1000));
    }
    
    @Override
    public void processElement(Order order, Context ctx, Collector<Order> out) {
        int score = 100;
        List<String> errors = new ArrayList<>();
        
        // 必填字段检查
        if (order.getOrderId() == null || order.getOrderId().isEmpty()) {
            errors.add("订单 ID 为空");
            score -= 20;
        }
        
        // 数值范围检查
        if (order.getAmount() == null || order.getAmount().compareTo(BigDecimal.ZERO) <= 0) {
            errors.add("订单金额异常");
            score -= 20;
        }
        
        // 枚举值检查
        if (!Arrays.asList("PENDING", "PAID", "CANCELLED").contains(order.getStatus())) {
            errors.add("订单状态非法");
            score -= 15;
        }
        
        // 时间合理性检查
        if (order.getCreateTime() == null || order.getCreateTime() > System.currentTimeMillis()) {
            errors.add("订单时间异常");
            score -= 15;
        }
        
        // 数据完整性检查
        if (order.getUserId() == null) {
            errors.add("用户 ID 为空");
            score -= 15;
        }
        
        qualityScore.update(score);
        
        if (score >= 80) {
            validCounter.inc();
            out.collect(order);
        } else {
            invalidCounter.inc();
            log.error("数据质量不合格: orderId={}, score={}, errors={}", 
                order.getOrderId(), score, errors);
            // 发送到错误队列
            ctx.output(errorOutputTag, order);
        }
    }
}

-- 查看 Binlog 配置
SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';

-- 开启 Binlog（需要重启）
-- my.cnf
[mysqld]
server-id=1
log-bin=mysql-bin
binlog_format=ROW
binlog_row_image=FULL
expire_logs_days=7

-- 创建 CDC 用户
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

-- 查看 Binlog 文件
SHOW BINARY LOGS;

-- 查看 Binlog 内容
SHOW BINLOG EVENTS IN 'mysql-bin.000001';