05. Apache Flink 实时计算详解

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class StreamingJob {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        // 1. 读取数据流（Source）
        DataStream<String> stream = env
            .fromSource(
                kafkaSource,
                WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5)),
                "Kafka Source"
            );

        // 2. 转换（Transformation）
        DataStream<OrderEvent> events = stream
            .map(new JsonToOrderMapper())
            .filter(e -> e.amount > 0);

        // 3. KeyBy + 窗口聚合
        DataStream<Tuple2<String, Double>> result = events
            .keyBy(e -> e.category)
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .aggregate(new SumAggregate());

        // 4. 输出（Sink）
        result.addSink(new FlinkJedisPoolConfig.Builder()
            .setHost("redis")
            .build());

        // 执行
        env.execute("RealTimeAnalytics");
    }
}

// KeyedProcessFunction：底层 API，可访问状态和定时器
class OrderTimeoutAlert extends KeyedProcessFunction<String, OrderEvent, Alert> {
    private ValueState<Long> timerState;

    @Override
    public void processElement(OrderEvent event, Context ctx, Collector<Alert> out) {
        long timeout = event.createTime + 15 * 60 * 1000; // 15分钟超时
        ctx.timerService().registerEventTimeTimer(timeout);
        timerState.update(timeout);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) {
        out.collect(new Alert(ctx.getCurrentKey(), "ORDER_TIMEOUT"));
    }
}

// Checkpoint 配置
env.enableCheckpointing(60000); // 60s 间隔
env.getCheckpointConfig().setCheckpointingMode(
    CheckpointingMode.EXACTLY_ONCE
);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setCheckpointTimeout(600000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().enableUnalignedCheckpoints();

// State Backend
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage("hdfs:///checkpoints");

// ValueState 示例：去重计数
class UniqueUserCount extends RichFlatMapFunction<Event, Metrics> {
    private ValueState<HashSet<String>> userState;

    @Override
    public void open(Configuration parameters) {
        StateTtlConfig ttl = StateTtlConfig
            .newBuilder(Time.hours(24))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
            .cleanupFullSnapshot()
            .build();

        userState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("users", Types.SET(Types.STRING))
        );
        // 启用 TTL
        ((ValueStateDescriptor<HashSet<String>>) userState.getStateDescriptor()).enableTimeToLive(ttl);
    }

    @Override
    public void flatMap(Event event, Collector<Metrics> out) throws Exception {
        HashSet<String> users = userState.value();
        if (users == null) users = new HashSet<>();
        users.add(event.userId);
        userState.update(users);
        out.collect(new Metrics(event.windowStart, users.size()));
    }
}

// RocksDB 增量 Checkpoint（推荐生产配置）
EmbeddedRocksDBStateBackend rocksDb = new EmbeddedRocksDBStateBackend(true);
env.setStateBackend(rocksDb);
env.getCheckpointConfig().setCheckpointStorage("hdfs:///flink-checkpoints");

// Event Time 配置
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

// Watermark 策略
stream.assignTimestampsAndWatermarks(
    WatermarkStrategy
        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(30))
        .withTimestampAssigner((event, timestamp) -> event.eventTime)
        .withIdleness(Duration.ofMinutes(5)) // 空闲源处理
);

// 自定义 Watermark 生成器
class PunctuatedWatermarkStrategy implements WatermarkStrategy<Event> {
    @Override
    public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context ctx) {
        return new WatermarkGenerator<Event>() {
            private long maxTimestamp = Long.MIN_VALUE;

            @Override
            public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
                maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
                // 特定事件触发 Watermark 推进
                if (event.isEndOfBatch) {
                    output.emitWatermark(new Watermark(maxTimestamp));
                }
            }

            @Override
            public void onPeriodicEmit(WatermarkOutput output) {
                output.emitWatermark(new Watermark(maxTimestamp - 5000));
            }
        };
    }
}

stream
    .keyBy(e -> e.userId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .allowedLateness(Time.minutes(2))    // 允许 2 分钟迟到
    .sideOutputLateData(lateTag)         // 超时的进侧输出流
    .aggregate(new CountAggregate());

// 处理侧输出迟到数据
DataStream<Event> lateStream = result.getSideOutput(lateTag);
lateStream.addSink(new LateDataSink()); // 写入死信队列/冷存储

// Tumbling Window：每 5 分钟统计一次
stream.keyBy(e -> e.category)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new AverageAggregate());

// Sliding Window：每 1 分钟输出过去 5 分钟的统计
stream.keyBy(e -> e.category)
    .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
    .aggregate(new SumAggregate());

// Session Window：10 分钟无活动则关闭会话
stream.keyBy(e -> e.userId)
    .window(EventTimeSessionWindows.withGap(Time.minutes(10)))
    .aggregate(new SessionAggregate());

// AggregateFunction：增量聚合，内存友好
class AverageAggregate implements AggregateFunction<Event, Tuple2<Long, Long>, Double> {
    @Override
    public Tuple2<Long, Long> createAccumulator() {
        return Tuple2.of(0L, 0L);
    }

    @Override
    public Tuple2<Long, Long> add(Event event, Tuple2<Long, Long> acc) {
        return Tuple2.of(acc.f0 + event.value, acc.f1 + 1);
    }

    @Override
    public Double getResult(Tuple2<Long, Long> acc) {
        return acc.f0 / (double) acc.f1;
    }

    @Override
    public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
        return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
    }
}

// ProcessWindowFunction：获取窗口元数据（全量聚合，内存消耗大）
class TopNProcess extends ProcessWindowFunction<Event, Result, String, TimeWindow> {
    @Override
    public void process(String key, Context ctx, Iterable<Event> events, Collector<Result> out) {
        long windowStart = ctx.window().getStart();
        List<Event> sorted = StreamSupport.stream(events.spliterator(), false)
            .sorted(Comparator.comparing(e -> -e.value))
            .limit(10)
            .collect(Collectors.toList());
        out.collect(new Result(key, windowStart, sorted));
    }
}

// Kafka Producer 配置：Exactly-Once Sink
Properties props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("transactional.id", "flink-producer-1"); // 事务 ID
props.put("enable.idempotence", "true");
props.put("acks", "all");

FlinkKafkaProducer<String> sink = new FlinkKafkaProducer<>(
    "output-topic",
    new SimpleStringSchema(),
    props,
    FlinkKafkaProducer.Semantic.EXACTLY_ONCE
);

stream.addSink(sink);

// At-Least-Once 配置（允许重复，性能更高）
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);

// Sink 配合幂等写入（如使用业务主键）
// 即使重复投递，最终结果一致

# YARN Session 模式
./bin/yarn-session.sh -n 4 -jm 2048 -tm 8192 -s 4

# Per-Job 模式
./bin/flink run -m yarn-cluster \
    -yn 4 -yjm 2048 -ytm 8192 \
    -ys 4 -p 16 \
    ./job.jar

# Kubernetes Native
./bin/flink run-application \
    --target kubernetes-application \
    -Dkubernetes.cluster-id=flink-cluster \
    -Dkubernetes.container.image=flink-job:latest \
    ./job.jar

// 监控反压
env.getConfig().setAutoWatermarkInterval(200);

// 反压常见原因与解决
// 1. 算子处理慢 → 增加并行度或优化逻辑
// 2. Sink 吞吐量不足 → 批量写入、异步化
// 3. 数据倾斜 → Key 加盐重新分区
// 4. GC 频繁 → 调整内存配置，使用 G1GC

// 生产推荐配置：RocksDB + 增量 Checkpoint
Configuration conf = new Configuration();

// 启用增量 Checkpoint
conf.setBoolean(StateBackendOptions.ASYNC_SNAPSHOTS, true);

EmbeddedRocksDBStateBackend rocksDb = new EmbeddedRocksDBStateBackend(true);
env.setStateBackend(rocksDb);
env.getCheckpointConfig().setCheckpointStorage("hdfs:///flink-checkpoints");

// RocksDB 内存调优
env.getConfig().set(ConfigConstants.ROCKSDB_MEMORY_MANAGED_KEY, "true");
env.getConfig().set(ConfigConstants.ROCKSDB_FIXED_PER_SLOT_MEMORY_SIZE, "256mb");

// 增量 Checkpoint 高级选项
env.getCheckpointConfig().enableUnalignedCheckpoints();
env.getCheckpointConfig().setAlignmentTimeout(Duration.ofSeconds(30));

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.hours(48))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
    .cleanupIncrementally(10, true)
    .build();

ValueStateDescriptor<Long> descriptor = new ValueStateDescriptor<>("counter", Types.LONG);
descriptor.enableTimeToLive(ttlConfig);

Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "kafka:9092");
producerProps.put("transactional.id", "flink-txn-sink-" + subtaskIndex);
producerProps.put("enable.idempotence", "true");
producerProps.put("acks", "all");
producerProps.put("retries", Integer.MAX_VALUE);
producerProps.put("max.in.flight.requests.per.connection", "5");

FlinkKafkaProducer<String> kafkaSink = new FlinkKafkaProducer<>(
    "output-events",
    new JsonSerializationSchema(),
    producerProps,
    FlinkKafkaProducer.Semantic.EXACTLY_ONCE
);

stream.addSink(kafkaSink);

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// Checkpoint 基础配置
env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setCheckpointTimeout(600000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

// 超时未对齐回退到 Aligned
env.getCheckpointConfig().enableUnalignedCheckpoints();
env.getCheckpointConfig().setAlignmentTimeout(Duration.ofSeconds(30));

// Source: Kafka 自动提交关闭，依赖 Checkpoint 保存 offset
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "kafka:9092");
consumerProps.put("group.id", "flink-exactly-once-group");
consumerProps.put("auto.offset.reset", "earliest");
consumerProps.put("enable.auto.commit", "false"); // 必须关闭

FlinkKafkaConsumer<Event> kafkaSource = new FlinkKafkaConsumer<>(
    "input-events",
    new EventDeserializationSchema(),
    consumerProps
);
kafkaSource.setCommitOffsetsOnCheckpoints(true);

DataStream<Event> source = env.addSource(kafkaSource);
source.addSink(kafkaSink);

env.execute("ExactlyOncePipeline");

Pattern<LoginEvent, LoginEvent> loginFailPattern = Pattern
    .<LoginEvent>begin("first")
        .where(evt -> evt.status.equals("FAIL"))
    .next("second")
        .where(evt -> evt.status.equals("FAIL"))
    .next("third")
        .where(evt -> evt.status.equals("FAIL"))
    .within(Time.minutes(5));

PatternStream<LoginEvent> patternStream = CEP.pattern(
    loginStream.keyBy(e -> e.userId),
    loginFailPattern
);

DataStream<Alert> alerts = patternStream.process(
    new PatternProcessFunction<LoginEvent, Alert>() {
        @Override
        public void processMatch(
            Map<String, List<LoginEvent>> match,
            Context ctx,
            Collector<Alert> out
        ) {
            LoginEvent first = match.get("first").get(0);
            out.collect(new Alert(
                first.userId,
                "TRIPLE_LOGIN_FAIL",
                ctx.timestamp()
            ));
        }
    }
);

Pattern<PaymentEvent, PaymentEvent> paymentPattern = Pattern
    .<PaymentEvent>begin("create")
        .where(evt -> evt.type.equals("CREATE"))
    .followedBy("pay")
        .where(evt -> evt.type.equals("PAY"))
    .within(Time.minutes(10));

PatternStream<PaymentEvent> ps = CEP.pattern(
    paymentStream.keyBy(e -> e.orderId),
    paymentPattern
);

// 处理超时：订单创建后 10 分钟未支付
OutputTag<String> timeoutTag = new OutputTag<String>("timeout") {};

SingleOutputStreamOperator<OrderResult> result = ps.process(
    new PatternTimeoutHandler<OrderResult, String>()
);

// 超时侧输出流
DataStream<String> timeoutStream = result.getSideOutput(timeoutTag);
timeoutStream.addSink(new OrderTimeoutSink());

// 场景：刷券套利检测 —— 同一设备在短时间内高频切换用户领取优惠券
Pattern<CouponEvent, CouponEvent> arbitragePattern = Pattern
    .<CouponEvent>begin("claim1")
        .where(evt -> evt.action.equals("CLAIM"))
    .next("claim2")
        .where(evt -> evt.action.equals("CLAIM"))
        .where(new SimpleCondition<CouponEvent>() {
            @Override
            public boolean filter(CouponEvent evt) {
                return !evt.userId.equals(prevUserId);
            }
        })
    .next("claim3")
        .where(evt -> evt.action.equals("CLAIM"))
    .within(Time.seconds(30));

// 匹配到模式后写入风控事件队列，触发人工审核
patternStream.process(new FraudAlertHandler()).addSink(new RiskControlSink());

-- 注册 Kafka Source 表
CREATE TABLE user_events (
    user_id STRING,
    event_type STRING,
    event_time TIMESTAMP(3),
    amount DECIMAL(10, 2),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'user-events',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'json',
    'scan.startup.mode' = 'latest-offset'
);

-- 注册 Upsert Kafka Sink 表
CREATE TABLE event_summary (
    event_type STRING PRIMARY KEY NOT ENFORCED,
    total_amount DECIMAL(18, 2),
    event_count BIGINT
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'event-summary',
    'properties.bootstrap.servers' = 'kafka:9092',
    'key.format' = 'json',
    'value.format' = 'json'
);

-- 流式聚合写入 Sink
INSERT INTO event_summary
SELECT event_type, SUM(amount) AS total_amount, COUNT(*) AS event_count
FROM user_events
GROUP BY event_type;

-- 流表
CREATE TABLE orders (
    order_id STRING,
    currency STRING,
    amount DECIMAL(10, 2),
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '10' SECOND
) WITH ('connector' = 'kafka', ...);

-- 维度表（MySQL CDC 或 Lookup）
CREATE TABLE currency_rates (
    currency STRING,
    rate DECIMAL(10, 6),
    update_time TIMESTAMP(3),
    PRIMARY KEY (currency) NOT ENFORCED,
    WATERMARK FOR update_time AS update_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://mysql:3306/warehouse',
    'table-name' = 'currency_rates',
    'lookup.cache.max-rows' = '1000',
    'lookup.cache.ttl' = '10 min'
);

-- Temporal Join：关联该订单时间点的最新汇率
SELECT
    o.order_id,
    o.amount,
    o.amount * r.rate AS amount_usd
FROM orders AS o
LEFT JOIN currency_rates FOR SYSTEM_TIME AS OF o.order_time AS r
ON o.currency = r.currency;

-- TUMBLE Window TVF
SELECT
    window_start,
    window_end,
    user_id,
    COUNT(*) AS event_count,
    SUM(amount) AS total_amount
FROM TABLE(
    TUMBLE(TABLE user_events, DESCRIPTOR(event_time), INTERVAL '10' MINUTES)
)
GROUP BY window_start, window_end, user_id;

-- CUMULATE（累积窗口）：每 1 分钟输出一次过去 1 小时的累积结果
SELECT
    window_start,
    window_end,
    event_type,
    COUNT(*) AS cnt
FROM TABLE(
    CUMULATE(
        TABLE user_events,
        DESCRIPTOR(event_time),
        INTERVAL '1' MINUTES,
        INTERVAL '1' HOUR
    )
)
GROUP BY window_start, window_end, event_type;

# 启动 Flink SQL Client
./bin/sql-client.sh embedded

# 或在会话中提交 SQL 文件
./bin/sql-client.sh -f /path/to/job.sql

# 设置执行模式
SET 'execution.runtime-mode' = 'streaming';
SET 'sql-client.execution.result-mode' = 'table';

# 查看执行计划
EXPLAIN PLAN FOR SELECT ... ;

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: realtime-etl
spec:
  image: flink:1.18-scala_2.12
  flinkVersion: v1.18
  jobManager:
    resource:
      memory: "2Gi"
      cpu: 1
  taskManager:
    resource:
      memory: "4Gi"
      cpu: 2
    replicas: 3
  job:
    jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
    parallelism: 6
    upgradeMode: stateful
    state: running

apiVersion: v1
kind: Pod
metadata:
  name: task-manager-template
spec:
  containers:
    - name: flink-main-container
      resources:
        requests:
          memory: "4Gi"
          cpu: "2"
        limits:
          memory: "4Gi"
          cpu: "2"
      env:
        - name: FLINK_TM_HEAP_SIZE
          value: "3072m"
        - name: JVM_ARGS
          value: "-XX:+UseG1GC -XX:MaxGCPauseMillis=100"
      volumeMounts:
        - name: checkpoint-volume
          mountPath: /checkpoints
  volumes:
    - name: checkpoint-volume
      persistentVolumeClaim:
        claimName: flink-checkpoint-pvc

# Flink Autoscaler（基于自定义指标或 CPU / 背压）
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: flink-taskmanager-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: flink-taskmanager
  minReplicas: 2
  maxReplicas: 20
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70
    - type: Pods
      pods:
        metric:
          name: flink_taskmanager_job_task_backPressuredTimeMsPerSecond
        target:
          type: AverageValue
          averageValue: "200"
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
        - type: Percent
          value: 100
          periodSeconds: 60

// 网络缓冲区配置（flink-conf.yaml）
// taskmanager.memory.network.fraction: 0.15
// taskmanager.memory.network.max: 256mb
// taskmanager.memory.network.min: 128mb

// 在代码中动态调整缓冲区
Configuration netConf = new Configuration();
netConf.set(TaskManagerOptions.NETWORK_MEMORY_FRACTION, 0.15);
netConf.set(TaskManagerOptions.NETWORK_MEMORY_MAX, MemorySize.parse("256mb"));
netConf.set(TaskManagerOptions.NETWORK_MEMORY_MIN, MemorySize.parse("128mb"));

// 启用 Credit-based 流量控制（默认已开启，Flink 1.5+）
// 解决反压的另一种思路：增大下游并行度或批量写入

// 开启对象链接重用（谨慎使用，仅当确认不会修改输入对象时）
env.getConfig().enableObjectReuse();

// 配合 ObjectReuse 的算子示例：避免在 FlatMap 中创建新对象
class OptimizedMapper extends RichFlatMapFunction<String, Event> {
    private final Event reuseEvent = new Event();

    @Override
    public void flatMap(String line, Collector<Event> out) {
        // 复用同一对象实例，减少 heap 分配
        reuseEvent.userId = parseUserId(line);
        reuseEvent.timestamp = parseTimestamp(line);
        reuseEvent.value = parseValue(line);
        out.collect(reuseEvent);
    }
}

// 设置 G1GC 降低 STW
// env.getConfig().addJVMOptions("-XX:+UseG1GC -XX:MaxGCPauseMillis=50");

// Pulsar Source
PulsarSource<String> pulsarSource = PulsarSource.builder()
    .setServiceUrl("pulsar://pulsar:6650")
    .setAdminUrl("http://pulsar:8080")
    .setStartCursor(StartCursor.earliest())
    .setTopics("persistent://public/default/events")
    .setDeserializationSchema(new SimpleStringSchema())
    .setSubscriptionName("flink-consumer-sub")
    .setSubscriptionType(SubscriptionType.Key_Shared)
    .build();

DataStream<String> stream = env.fromSource(
    pulsarSource,
    WatermarkStrategy.noWatermarks(),
    "Pulsar Source"
);

// Pulsar Sink
PulsarSink<String> pulsarSink = PulsarSink.builder()
    .setServiceUrl("pulsar://pulsar:6650")
    .setAdminUrl("http://pulsar:8080")
    .setProducerConfig(
        PulsarSinkOptions.PULSAR_BATCHING_ENABLED,
        Boolean.TRUE
    )
    .setTopics("persistent://public/default/output")
    .setSerializationSchema(new SimpleStringSchema())
    .build();

stream.sinkTo(pulsarSink);

// Key-Shared 订阅确保顺序性与并行扩展同时满足
PulsarSource<Event> keyedSource = PulsarSource.builder()
    .setServiceUrl("pulsar://pulsar:6650")
    .setTopicsList(Arrays.asList("order-events"))
    .setDeserializationSchema(new EventSchema())
    .setSubscriptionName("flink-key-shared-sub")
    .setSubscriptionType(SubscriptionType.Key_Shared)
    // 允许粘性 Key 散列范围重新分配
    .setConfig(PulsarSourceOptions.PULSAR_ALLOW_TOPIC_LISTENER, true)
    .build();

// Pulsar Schema 自动注册与演进
Schema<Event> eventSchema = Schema.AVRO(Event.class);

PulsarSource<Event> sourceWithSchema = PulsarSource.builder()
    .setServiceUrl("pulsar://pulsar:6650")
    .setTopics("events-topic")
    .setSchema(eventSchema)
    .setSubscriptionName("flink-schema-sub")
    .build();

// Schema 兼容性策略在 Pulsar 侧配置：FULL / BACKWARD / FORWARD

// 诊断性配置：输出 Checkpoint 统计到日志
env.getCheckpointConfig().enableCheckpointingIntervalLogging();

// 若 Checkpoint 超时，临时增大 timeout 并启用增量
checkpointConfig.setCheckpointTimeout(900000);
((EmbeddedRocksDBStateBackend) env.getStateBackend()).setIncrementalRestorePath(...);

// 方案 1：两阶段聚合（Local + Global）
DataStream<Result> preAggregated = stream
    .map(e -> Tuple2.of(randomPrefix(e.userId, 10), e.amount))
    .keyBy(t -> t.f0)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new PreAggregate())
    .map(t -> Tuple2.of(removePrefix(t.f0), t.f1))
    .keyBy(t -> t.f0)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new FinalAggregate());

// 方案 2：自定义分区器打散热点 Key
stream.partitionCustom(new HotKeyPartitioner(), e -> e.userId);