3、精算数据流接入：Kafka数据源配置、JSON/AVRO格式解析、Schema Registry集成、数据质量校验

精算坏样本归因的首要前提是稳定、可靠地接入上游业务系统的保单、理赔、核保等数据。本章聚焦于如何利用 Flink 连接 Kafka 作为数据源，处理 JSON 与 AVRO 两种主流序列化格式，集成 Schema Registry 实现模式管理，并在数据进入计算管道前完成质量校验。

3.1 Kafka 数据源配置

Flink 通过 FlinkKafkaConsumer（旧版 API）或 KafkaSource（新版 API）接入 Kafka 主题。精算场景下，建议使用 KafkaSource 以获得更好的 exactly-once 语义与动态分区发现能力。

// Flink 1.17+ KafkaSource 配置示例
KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("kafka1:9092,kafka2:9092")
    .setTopics("actuary.claim.raw", "actuary.policy.raw")
    .setGroupId("flink-actuary-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setProperties(immutableMap.of(
        "enable.auto.commit", "false",
        "partition.discovery.interval.ms", "60000"
    ))
    .build();

3.2 JSON 与 AVRO 格式解析

精算数据流中，JSON 常用于非结构化字段（如理赔描述、核保备注），AVRO 则用于结构化保单、费率表等强类型数据。Flink 需要根据数据格式选择对应的反序列化器。

3.2.1 JSON 解析

使用 SimpleStringSchema 接收原始字符串后，通过 JsonDeserializationSchema 或自定义 ObjectMapper 转换为 POJO。精算字段中常见嵌套 JSON，建议使用 JSONPath 提取关键指标。

// 自定义 JSON 反序列化
DataStream<ClaimEvent> claimStream = env
    .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source")
    .map(jsonStr -> {
        ObjectMapper mapper = new ObjectMapper();
        JsonNode root = mapper.readTree(jsonStr);
        return new ClaimEvent(
            root.path("claimId").asText(),
            root.path("policyNo").asText(),
            root.path("lossAmount").asDouble(),
            root.path("occurDate").asText()
        );
    });

3.2.2 AVRO 解析

AVRO 需要预定义 Schema，Flink 提供 AvroDeserializationSchema 或 ConfluentRegistryAvroDeserializationSchema。精算团队应提前将 AVRO Schema 注册到 Schema Registry，避免硬编码。

// AVRO 反序列化（使用静态 Schema）
Schema schema = new Schema.Parser().parse(
    getClass().getResourceAsStream("/avro/PolicyEvent.avsc")
);
DataStream<PolicyEvent> policyStream = env
    .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "AVRO Source")
    .map(new AvroDeserializationSchema<PolicyEvent>(PolicyEvent.class));

3.3 Schema Registry 集成

Schema Registry 是精算数据治理的核心组件，确保生产者和消费者使用兼容的 Schema。Flink 通过 Confluent 提供的 KafkaAvroDeserializer 集成 Registry，支持向前/向后兼容。

// 集成 Schema Registry 的 AVRO 反序列化
Properties props = new Properties();
props.setProperty("bootstrap.servers", "kafka1:9092");
props.setProperty("schema.registry.url", "http://schema-registry:8081");
props.setProperty("specific.avro.reader", "true");

KafkaSource<PolicyEvent> avroSource = KafkaSource.<PolicyEvent>builder()
    .setBootstrapServers("kafka1:9092")
    .setTopics("actuary.policy.avro")
    .setGroupId("flink-actuary-group")
    .setDeserializer(
        ConfluentRegistryAvroDeserializationSchema.forSpecific(PolicyEvent.class, PolicyEvent.class, props)
    )
    .build();

3.4 数据质量校验

坏样本归因对数据质量高度敏感——缺失的保额、异常的赔付率、错误的日期格式都会导致归因偏差。Flink 管道中应在 source 之后立即嵌入质量校验算子。

3.4.1 校验维度

• 完整性： 关键字段（claimId、policyNo、lossAmount）非空
• 准确性： 数值范围校验（如 lossAmount > 0 且 < 1e8）
• 一致性： 日期格式匹配 yyyy-MM-dd，枚举值在允许列表中
• 时效性： 事件时间戳不能晚于当前时间 + 1 小时（允许时钟偏差）

3.4.2 校验实现

// 数据质量校验算子示例
DataStream<ClaimEvent> validatedStream = claimStream
    .filter(event -> {
        if (event.getClaimId() == null || event.getClaimId().isEmpty()) {
            Metrics.counter("dq.missing.claimId").inc();
            return false;
        }
        if (event.getLossAmount() <= 0 || event.getLossAmount() > 1e8) {
            Metrics.counter("dq.invalid.lossAmount").inc();
            return false;
        }
        // 日期格式校验
        try {
            LocalDate.parse(event.getOccurDate(), DateTimeFormatter.ISO_LOCAL_DATE);
        } catch (Exception e) {
            Metrics.counter("dq.bad.date").inc();
            return false;
        }
        return true;
    })
    .name("DataQualityFilter");

3.4.3 异常数据路由

校验失败的数据不应直接丢弃，应路由到侧输出（Side Output）或死信队列（DLQ），供精算团队事后分析。

// 侧输出收集异常数据
OutputTag<ClaimEvent> dqTag = new OutputTag<ClaimEvent>("dq-failed") {};
SingleOutputStreamOperator<ClaimEvent> mainStream = validatedStream
    .process(new ProcessFunction<ClaimEvent, ClaimEvent>() {
        @Override
        public void processElement(ClaimEvent value, Context ctx, Collector<ClaimEvent> out) {
            if (passesAllChecks(value)) {
                out.collect(value);
            } else {
                ctx.output(dqTag, value);
            }
        }
    });

DataStream<ClaimEvent> dqFailedStream = mainStream.getSideOutput(dqTag);
dqFailedStream.addSink(new KafkaSink<>(...)); // 写入 DLQ 主题

3.5 本章小结