2. 交易流水的实时清洗与字段标准化

任务目标

从 Kafka 读取原始交易流水（JSON 字符串），解析出交易字段，完成类型转换、空值过滤与格式统一，输出标准化交易事件流。标准化规则：金额转 double 且必须 > 0；时间戳统一为毫秒 long；币种统一大写；交易类型统一小写；txId、accountId 任一为空则丢弃。

环境与依赖

• Flink 1.17+，Kafka 连接器 flink-connector-kafka。
• Kafka 中存在 topic raw_tx，消息为 JSON 字符串。
• Maven 依赖：flink-streaming-java、flink-connector-kafka、flink-json（或 Jackson）。

处理链路

核心实现

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
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;

public class TxCleanJob {

    // 标准化交易事件
    public static class TxEvent {
        public String txId;
        public String accountId;
        public double amount;
        public String currency;   // 统一大写
        public String txType;     // 统一小写
        public long ts;           // 毫秒

        public TxEvent() {}
        public TxEvent(String txId, String accountId, double amount,
                       String currency, String txType, long ts) {
            this.txId = txId; this.accountId = accountId; this.amount = amount;
            this.currency = currency; this.txType = txType; this.ts = ts;
        }
        @Override public String toString() {
            return txId + "," + accountId + "," + amount + "," + currency + "," + txType + "," + ts;
        }
    }

    private static final ObjectMapper MAPPER = new ObjectMapper();

    // 解析 + 清洗 + 标准化；非法数据返回 null
    public static TxEvent clean(String raw) {
        try {
            JsonNode n = MAPPER.readTree(raw);
            String txId = text(n, "txId");
            String accountId = text(n, "accountId");
            if (txId == null || txId.isEmpty() || accountId == null || accountId.isEmpty()) {
                return null; // 关键字段空值过滤
            }
            double amount = n.path("amount").asDouble(0.0);
            if (amount <= 0) return null; // 金额非法过滤

            String currency = text(n, "currency");
            if (currency == null) return null;
            currency = currency.trim().toUpperCase();

            String txType = text(n, "txType");
            if (txType == null) return null;
            txType = txType.trim().toLowerCase();

            long ts = n.path("ts").asLong(0L);
            if (ts <= 0) return null;

            return new TxEvent(txId, accountId, amount, currency, txType, ts);
        } catch (Exception e) {
            return null; // 解析失败丢弃
        }
    }

    private static String text(JsonNode n, String field) {
        JsonNode v = n.get(field);
        return v == null || v.isNull() ? null : v.asText();
    }

    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("raw_tx")
                .setGroupId("tx-clean-group")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

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

        DataStream<TxEvent> cleaned = raw
                .map(TxCleanJob::clean)
                .filter(e -> e != null);

        cleaned.print("std-tx");
        env.execute("tx-clean-job");
    }
}

运行与验证

1. 启动 Kafka，创建 topic：kafka-topics.sh --create --topic raw_tx --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1。
2. 运行 TxCleanJob.main。
3. 向 topic 发送测试消息：

   {"txId":"T1","accountId":"A1","amount":"100.5","currency":"cny","txType":"PAY","ts":1700000000000}
   {"txId":"","accountId":"A2","amount":50,"currency":"USD","txType":"pay","ts":1700000000001}
   {"txId":"T3","accountId":"A3","amount":-5,"currency":"USD","txType":"pay","ts":1700000000002}
   {"txId":"T4","accountId":"A4","amount":20,"currency":"usd","txType":"REFUND","ts":1700000000003}

4. 观察控制台 std-tx 输出。通过条件：仅 T1、T4 被输出，且 T1 为 T1,A1,100.5,CNY,pay,1700000000000，T4 为 T4,A4,20.0,USD,refund,1700000000003；空 txId 与负金额记录被过滤。

说明：amount 为字符串 "100.5" 时 Jackson 的 asDouble 可正常转换；若上游字段名或类型不同，需同步调整 clean 中的取值逻辑。