1. Flink 读取 Kafka 实时交易流水

任务目标

用 Flink DataStream API 从 Kafka 消费实时交易流水（JSON 字符串），解析出交易字段后打印到控制台，验证「Kafka → Flink → 控制台」接入链路是否打通。

环境与依赖

• JDK 8/11，Maven；本地或容器可访问的 Kafka（默认 localhost:9092）。
• 依赖：flink-streaming-java、flink-clients、flink-connector-kafka、flink-json（Flink 1.17.x）。
• Kafka 中需存在 topic trade_flow，消息为 JSON 交易流水。

数据链路

核心代码

单文件即可运行。用 KafkaSource 接入，SimpleStringSchema 读原始 JSON，再用 map 解析为 Trade 并打印。

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
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.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;

public class KafkaTradeJob {

    // 交易流水实体
    public static class Trade {
        public String tradeId;
        public String account;
        public double amount;
        public long ts;

        public Trade() {}

        public Trade(String tradeId, String account, double amount, long ts) {
            this.tradeId = tradeId;
            this.account = account;
            this.amount = amount;
            this.ts = ts;
        }

        @Override
        public String toString() {
            return "Trade{tradeId='" + tradeId + "', account='" + account
                    + "', amount=" + amount + ", ts=" + ts + "}";
        }
    }

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

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092")
                .setTopics("trade_flow")
                .setGroupId("flink-trade-group")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStream<String> raw = env.fromSource(
                source, WatermarkStrategy.noWatermarks(), "kafka-trade-source");

        ObjectMapper mapper = new ObjectMapper();

        DataStream<Trade> trades = raw.map(json -> {
            JsonNode node = mapper.readTree(json);
            return new Trade(
                    node.path("tradeId").asText(),
                    node.path("account").asText(),
                    node.path("amount").asDouble(),
                    node.path("ts").asLong());
        });

        trades.print("TRADE");

        env.execute("Flink Read Kafka Trade Flow");
    }
}

运行方式

# 1. 启动 Kafka 并创建 topic
bin/kafka-topics.sh --create --topic trade_flow \
  --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

# 2. 编译打包
mvn -q clean package

# 3. 提交运行（本地模式）
flink run -c KafkaTradeJob target/flink-trade-1.0.jar

验证方法与通过条件

1. 作业启动后，Flink Web UI 或日志中应出现算子 kafka-trade-source 与 TRADE 打印算子，且处于 RUNNING。
2. 向 topic 发送一条测试消息：

   echo '{"tradeId":"T1001","account":"A001","amount":2500.5,"ts":1700000000000}' \
     | bin/kafka-console-producer.sh --topic trade_flow --bootstrap-server localhost:9092

3. TaskManager 日志中应出现形如 TRADE> Trade{tradeId='T1001', account='A001', amount=2500.5, ts=1700000000000} 的输出行。
4. 通过条件：发送的每条合法 JSON 都能在控制台看到对应解析后的 Trade 记录，字段值与发送内容一致，即链路打通。

常见问题

• 无输出：确认 setStartingOffsets 为 latest() 时消息是在作业启动后发送的；若想消费历史数据改为 earliest()。
• 解析报错：消息必须是合法 JSON，且字段名与代码中 tradeId/account/amount/ts 一致。
• 连接失败：检查 bootstrapServers 与 Kafka 实际地址、网络是否可达。