3. 数据源设计与模拟：预警数据源类型与 Flink DataGenerator 模拟流数据

实时预警系统的核心在于对持续流入的数据进行毫秒级分析。本章聚焦于如何设计并模拟三类典型的预警数据源——日志、传感器、交易，并基于 Flink 内置的 DataGenerator 工具生成高仿真流数据，为后续预警实验提供稳定、可控的输入。

3.1 预警数据源类型与特征对比

不同业务场景下的预警数据源在格式、速率、语义上差异显著。下表从关键维度对比三类数据源，帮助实验设计者选择合适的模拟策略。

3.2 使用 Flink DataGenerator 模拟流数据

Flink 的 DataGeneratorSource（自 1.17 起稳定）允许用户通过实现 GeneratorFunction 接口，以编程方式生成无界流数据。相比读取文件或 Kafka，它具备以下优势：

• 无外部依赖：无需启动 Kafka、文件系统等中间件，适合单元测试与快速原型。
• 速率可控：支持设置每秒生成记录数（recordsPerSecond），模拟不同负载。
• 数据可重复：基于种子（seed）的随机生成，保证实验可复现。

3.3 三类数据源的 DataGenerator 实现

以下代码片段分别展示日志、传感器、交易数据源的生成逻辑。所有示例均使用 Flink 1.18+ API。

3.3.1 日志数据源模拟

// 日志事件 POJO
public class LogEvent {
    public long timestamp;
    public String level;      // INFO, WARN, ERROR
    public String service;
    public int errorCode;
}

// GeneratorFunction 实现
DataGeneratorSource<LogEvent> logSource = new DataGeneratorSource<>(
    (GeneratorFunction<Long, LogEvent>) index -> {
        LogEvent event = new LogEvent();
        event.timestamp = System.currentTimeMillis();
        event.level = index % 10 == 0 ? "ERROR" : "INFO";
        event.service = "service-" + (index % 5);
        event.errorCode = event.level.equals("ERROR") ? 500 : 200;
        return event;
    },
    1000,   // 每秒生成 1000 条
    Types.POJO(LogEvent.class)
);

3.3.2 传感器数据源模拟

// 传感器读数 POJO
public class SensorReading {
    public String deviceId;
    public double temperature;
    public double humidity;
    public long ts;
}

DataGeneratorSource<SensorReading> sensorSource = new DataGeneratorSource<>(
    (GeneratorFunction<Long, SensorReading>) index -> {
        SensorReading reading = new SensorReading();
        reading.deviceId = "sensor-" + (index % 100);
        reading.temperature = 20.0 + (index % 50) * 0.5;  // 20°C ~ 45°C
        reading.humidity = 40.0 + (index % 30) * 1.0;     // 40% ~ 70%
        reading.ts = System.currentTimeMillis();
        return reading;
    },
    500,    // 每秒 500 条
    Types.POJO(SensorReading.class)
);

3.3.3 交易数据源模拟

// 交易事件 POJO
public class TransactionEvent {
    public String userId;
    public double amount;
    public String merchant;
    public long timestamp;
}

DataGeneratorSource<TransactionEvent> txSource = new DataGeneratorSource<>(
    (GeneratorFunction<Long, TransactionEvent>) index -> {
        TransactionEvent tx = new TransactionEvent();
        tx.userId = "user-" + (index % 1000);
        tx.amount = 50.0 + (index % 200) * 10.0;  // 50 ~ 2050 元
        tx.merchant = (index % 3 == 0) ? "Amazon" : "Shopify";
        tx.timestamp = System.currentTimeMillis();
        return tx;
    },
    200,    // 每秒 200 笔
    Types.POJO(TransactionEvent.class)
);

3.4 数据源模拟架构图

下图展示三类数据源通过 Flink DataGenerator 注入作业，经预处理后进入预警规则引擎的完整链路。

3.5 模拟数据质量检查清单

在将模拟数据接入预警实验前，建议逐项核对以下质量属性：

• 字段完整性：所有预警规则依赖的字段是否均已生成？例如传感器数据必须包含 deviceId 与 temperature。
• 时间戳合理性：事件时间是否单调递增？是否包含乱序数据（用于测试 watermark 机制）？
• 异常注入比例：是否按实验设计注入了特定比例的异常事件（如 5% 的 ERROR 日志）？
• 速率稳定性：使用 DataGeneratorSource 的背压感知模式，确保生成速率与下游处理能力匹配。

3.6 实验参数配置表

以下参数可在实验启动前通过 Configuration 或命令行动态调整，便于进行敏感性分析。

通过上述设计与实现，实验者可以快速搭建一个贴近真实业务的多源数据环境，为后续预警规则评估与性能基准测试奠定坚实的数据基础。