4. 每分钟交易笔数与交易金额实时统计

任务目标

从 Kafka 读取交易流水，按事件时间划分 1 分钟滚动窗口，统计每分钟的交易笔数与总交易金额，并将结果写入 Kafka 结果主题（同时给出写数据库的接入点）。

环境与依赖

• Flink 1.17+，Kafka 2.x+，JDK 11
• Maven 依赖：flink-streaming-java、flink-connector-kafka、flink-clients
• 输入主题 txn：JSON 格式 {"txnId":"t1","amount":88.5,"ts":1700000000000}
• 输出主题 txn_stat：JSON 格式 {"windowStart":...,"windowEnd":...,"count":...,"totalAmount":...}

处理链路

核心代码

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kafka.sink.KafkaSink;
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;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

public class MinuteTxnStat {
    static final ObjectMapper M = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092")
                .setTopics("txn")
                .setGroupId("minute-stat")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStream<String> raw = env.fromSource(source,
                WatermarkStrategy.<String>forBoundedOutOfOrderness(java.time.Duration.ofSeconds(5))
                        .withTimestampAssigner((e, t) -> parseTs(e)),
                "kafka-txn");

        DataStream<String> result = raw
                .map(MinuteTxnStat::toTxn)
                .keyBy(t -> "all")
                .window(TumblingEventTimeWindows.of(Time.minutes(1)))
                .aggregate(new TxnAgg())
                .map(MinuteTxnStat::toJson);

        KafkaSink<String> sink = KafkaSink.<String>builder()
                .setBootstrapServers("localhost:9092")
                .setRecordSerializer(new org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder<>()
                        .setTopic("txn_stat")
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build())
                .build();

        result.sinkTo(sink);
        env.execute("minute-txn-stat");
    }

    static long parseTs(String json) {
        try { return M.readTree(json).get("ts").asLong(); }
        catch (Exception e) { return System.currentTimeMillis(); }
    }

    static Txn toTxn(String json) throws Exception {
        JsonNode n = M.readTree(json);
        return new Txn(n.get("txnId").asText(), n.get("amount").asDouble(), n.get("ts").asLong());
    }

    static String toJson(Stat s) throws Exception {
        return M.writeValueAsString(s);
    }

    public static class Txn {
        public String txnId; public double amount; public long ts;
        public Txn() {}
        public Txn(String id, double a, long t) { txnId = id; amount = a; ts = t; }
    }

    public static class Stat {
        public long windowStart, windowEnd, count; public double totalAmount;
        public Stat() {}
        public Stat(long s, long e, long c, double a) { windowStart = s; windowEnd = e; count = c; totalAmount = a; }
    }

    public static class TxnAgg implements org.apache.flink.api.common.functions.AggregateFunction<Txn, Stat, Stat> {
        public Stat createAccumulator() { return new Stat(0, 0, 0, 0); }
        public Stat add(Txn t, Stat acc) {
            acc.count += 1;
            acc.totalAmount += t.amount;
            return acc;
        }
        public Stat getResult(Stat acc) { return acc; }
        public Stat merge(Stat a, Stat b) {
            return new Stat(0, 0, a.count + b.count, a.totalAmount + b.totalAmount);
        }
    }
}

运行与验证

1. 启动 Kafka，创建主题：kafka-topics.sh --create --topic txn --partitions 2 --replication-factor 1 --bootstrap-server localhost:9092，同样创建 txn_stat。
2. 运行 MinuteTxnStat（本地 IDE 或 flink run）。
3. 向 txn 发送若干条带 ts 的交易 JSON，例如同一分钟内 3 条金额 10、20、30。
4. 消费 txn_stat：kafka-console-consumer.sh --topic txn_stat --from-beginning --bootstrap-server localhost:9092。
5. 通过条件：窗口触发后输出一条记录，count=3、totalAmount=60.0，且 windowEnd - windowStart = 60000。

写入数据库的接入点

将 result.sinkTo(sink) 替换为自定义 RichSinkFunction<String>，在 invoke 中解析 JSON 并执行 INSERT INTO minute_stat(window_start,window_end,cnt,total_amount) VALUES(?,?,?,?)，用 JDBC 连接池即可，无需改动上游窗口逻辑。