// Event Time：事件发生时间（最常用，但需要处理乱序）
// Processing Time：处理时间（最简单，但结果不可复现）
// Ingestion Time：进入 Flink 的时间（折中方案）

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

// 假设数据最多乱序 5 秒
WatermarkStrategy
    .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withTimestampAssigner((event, timestamp) -> event.getTimestamp());

// 滚动窗口：每 5 分钟一个窗口，窗口不重叠
stream.keyBy(e -> e.userId)
      .window(TumblingEventTimeWindows.of(Time.minutes(5)))
      .sum("amount");

// 滑动窗口：窗口大小 10 分钟，每 5 分钟滑动一次
stream.keyBy(e -> e.userId)
      .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(5)))
      .sum("amount");

// 会话窗口：不活跃超过 30 分钟，窗口关闭
stream.keyBy(e -> e.userId)
      .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
      .sum("amount");

// Keyed State：每个 Key 独立的状态
public class CountFunction extends KeyedProcessFunction<String, Event, Result> {
    // 值状态：存储一个值
    private ValueState<Long> countState;

    // 列表状态：存储一个列表
    private ListState<Event> historyState;

    // Map状态：存储一个Map
    private MapState<String, Long> detailState;

    @Override
    public void open(Configuration parameters) {
        countState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("count", Long.class));
    }

    @Override
    public void processElement(Event event, Context ctx, Collector<Result> out) {
        Long count = countState.value();
        if (count == null) count = 0L;
        count++;
        countState.update(count);
        // ...
    }
}

// 启用 Checkpoint，每 60 秒一次
env.enableCheckpointing(60000);

// Exactly-Once 语义（更安全，但更慢）
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// At-Least-Once 语义（更快，但可能重复）
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);

-- 创建 Kafka 源表
CREATE TABLE order_source (
    order_id STRING,
    user_id STRING,
    amount DECIMAL(10,2),
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json'
);

-- 实时聚合：每分钟的订单统计
SELECT
    TUMBLE_START(order_time, INTERVAL '1' MINUTE) as window_start,
    COUNT(*) as order_cnt,
    SUM(amount) as total_amount
FROM order_source
GROUP BY TUMBLE(order_time, INTERVAL '1' MINUTE);

Copy-on-Write (COW)：
- 写入时复制整个文件
- 读取性能好（直接读 Parquet）
- 写入性能差（要重写文件）
- 适合读多写少的场景

Merge-on-Read (MOR)：
- 写入时只追加 Delta 文件
- 写入性能好
- 读取时需要合并（读性能略差）
- 适合写多读少的场景

-- Hudi 表创建示例
CREATE TABLE hudi_order (
    order_id STRING,
    user_id STRING,
    amount DECIMAL(10,2),
    status STRING,
    update_time TIMESTAMP
) USING hudi
OPTIONS (
    'primaryKey' = 'order_id',
    'type' = 'cow',
    'preCombineField' = 'update_time'
);

-- 支持 UPSERT（有则更新，无则插入）
MERGE INTO hudi_order target
USING source_data source
ON target.order_id = source.order_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

-- 时间旅行：查询昨天的数据快照
SELECT * FROM hudi_order TIMESTAMP AS OF '2024-06-14';

传统架构：
数据源 → 数据湖（原始存储） → 数据仓库（分析）
         ↑ 两套系统，数据要搬来搬去

湖仓一体：
数据源 → 数据湖 + 仓库能力（一套系统搞定）
         ↑ 存储和计算分离，同一份数据支持批/流/交互式分析

一个 Spark SQL 的执行过程：

SQL 语句
    ↓ 解析
逻辑计划（Logical Plan）
    ↓ 优化器（Catalyst）
优化后的逻辑计划
    ↓ 物理计划生成
物理计划（Physical Plan）
    ↓ 代码生成（Codegen）
RDD 执行图
    ↓ DAGScheduler
Stage 划分（以 Shuffle 为边界）
    ↓ TaskScheduler
Task 分发到 Executor 执行

-- 不好的写法：两次 Shuffle
SELECT a.user_id, b.order_cnt, c.pay_amount
FROM users a
JOIN (
    SELECT user_id, COUNT(*) as order_cnt
    FROM orders
    GROUP BY user_id
) b ON a.user_id = b.user_id
JOIN (
    SELECT user_id, SUM(amount) as pay_amount
    FROM payments
    GROUP BY user_id
) c ON a.user_id = c.user_id;

-- 优化后：合并子查询，减少 Shuffle
SELECT
    a.user_id,
    COUNT(DISTINCT o.order_id) as order_cnt,
    SUM(p.amount) as pay_amount
FROM users a
LEFT JOIN orders o ON a.user_id = o.user_id
LEFT JOIN payments p ON a.user_id = p.user_id
GROUP BY a.user_id;

-- 不好的写法：全表扫描
SELECT * FROM orders WHERE order_date >= '2024-06-01';

-- 好的写法：如果 dt 是分区字段，只扫描需要的分区
SELECT * FROM orders WHERE dt >= '2024-06-01';

-- 危险的写法：笛卡尔积
SELECT a.*, b.*
FROM table_a a
JOIN table_b b
ON a.key = b.key AND a.key IS NULL;
-- 如果 a.key 有很多 NULL，会产生笛卡尔积

-- 更危险的写法：CROSS JOIN
SELECT * FROM table_a CROSS JOIN table_b;
-- 1万行 x 1万行 = 1亿行

-- 启用 AQE
SET spark.sql.adaptive.enabled = true;

-- 自动合并小分区（避免大量小文件）
SET spark.sql.adaptive.coalescePartitions.enabled = true;

-- 自动处理数据倾斜
SET spark.sql.adaptive.skewJoin.enabled = true;

# Executor 内存配置
--executor-memory 8g
--conf spark.executor.memoryOverhead=2g

# 内存管理
--conf spark.memory.fraction=0.6      # 执行+存储内存占比
--conf spark.memory.storageFraction=0.5  # 存储内存占比

# GC 优化
--conf spark.executor.extraJavaOptions="-XX:+UseG1GC -XX:MaxGCPauseMillis=200"

-- 数据质量检查示例

-- 完整性检查：关键字段 NULL 比例
SELECT
    COUNT(*) as total,
    SUM(CASE WHEN user_id IS NULL THEN 1 ELSE 0 END) as null_cnt,
    SUM(CASE WHEN user_id IS NULL THEN 1 ELSE 0 END) / COUNT(*) as null_ratio
FROM dwd_order_detail
WHERE dt = '${bizdate}';

-- 一致性检查：订单金额和支付金额是否匹配
SELECT
    SUM(order_amount) as order_sum,
    SUM(pay_amount) as pay_sum,
    ABS(SUM(order_amount) - SUM(pay_amount)) / SUM(order_amount) as diff_ratio
FROM ads_daily_summary
WHERE dt = '${bizdate}';

-- 唯一性检查：主键是否重复
SELECT order_id, COUNT(*) as cnt
FROM dwd_order_detail
WHERE dt = '${bizdate}'
GROUP BY order_id
HAVING cnt > 1;

数据血缘的价值：
1. 影响分析：修改一张表前，知道会影响哪些下游
2. 问题追溯：数据错了，能快速定位是哪个环节出问题
3. 口径统一：知道每个指标是怎么算出来的

成本优化的几个方向：
1. 资源利用率：任务申请 100G 内存，实际只用 20G
2. 存储优化：历史数据压缩、冷热分层
3. 计算优化：避免重复计算，合理设置任务周期
4. 淘汰无用数据：很多表几个月没人用了，占着资源

基本概念（必须知道）：
- Pod：K8s 最小调度单位
- Deployment：管理 Pod 副本
- Service：服务发现和负载均衡
- ConfigMap/Secret：配置管理

实操技能（按需学习）：
- 能看懂 Spark/Flink on K8s 的 YAML 配置
- 能用 kubectl 查看日志、排查问题
- 理解 Spark on K8s 和 Spark on YARN 的区别

Lambda 架构：
     数据源
        ↓
   ┌────┴────┐
批处理层   实时处理层
   └────┬────┘
        ↓
     服务层

优点：批处理保证准确性，实时满足时效性
缺点：两套代码，维护成本高

Kappa 架构：
数据源 → 消息队列 → 实时处理 → 服务层
            ↑
        重放（回溯）

优点：一套代码，架构简单
缺点：对实时引擎要求高，历史重算成本高