📚 课程目录第3章数据源设计

3. 数据源设计：模拟实时点击流数据

实时坏样本归因分析的第一步，是构建一个可靠且可控的数据源。在生产环境中，点击流数据通常来自前端埋点、消息队列（如 Kafka）或日志采集系统。但在开发与测试阶段，我们需要一个可复现、可调节的模拟数据源，以便验证后续的 Flink 处理逻辑。

本章将完成三项核心任务：定义点击事件的数据结构（POJO）、设计模拟数据生成逻辑、使用 DataStreamSource 接入 Flink 流处理环境。通过本章的学习，你将掌握 Flink 中自定义数据源的标准方法，并为后续的坏样本识别与归因分析打下数据基础。

3.1 点击事件 POJO 定义

在 Flink 中，POJO（Plain Old Java Object）是最常用的数据类型之一。Flink 对 POJO 有明确的约束：必须包含无参构造器、所有字段可访问（public 或通过 getter/setter）、且字段类型可序列化。我们定义一个 ClickEvent 类，用于承载单次点击行为的所有属性。

字段名	类型	说明	示例值
userId	String	用户唯一标识	user_1024
eventId	String	事件唯一 ID（UUID）	evt_8a3f2c
pageId	String	页面 ID	page_detail
itemId	String	商品/内容 ID	item_5678
clickType	String	点击类型	add_cart / purchase / view
timestamp	long	事件发生时间戳（毫秒）	1710000000000
isBadSample	boolean	是否为坏样本（模拟标记）	false

// ClickEvent.java
public class ClickEvent {
    public String userId;
    public String eventId;
    public String pageId;
    public String itemId;
    public String clickType;
    public long timestamp;
    public boolean isBadSample;

    // 无参构造器（Flink 要求）
    public ClickEvent() {}

    // 全参构造器（方便快速创建）
    public ClickEvent(String userId, String eventId, String pageId,
                      String itemId, String clickType, long timestamp,
                      boolean isBadSample) {
        this.userId = userId;
        this.eventId = eventId;
        this.pageId = pageId;
        this.itemId = itemId;
        this.clickType = clickType;
        this.timestamp = timestamp;
        this.isBadSample = isBadSample;
    }

    @Override
    public String toString() {
        return "ClickEvent{" +
                "userId='" + userId + '\'' +
                ", eventId='" + eventId + '\'' +
                ", pageId='" + pageId + '\'' +
                ", itemId='" + itemId + '\'' +
                ", clickType='" + clickType + '\'' +
                ", timestamp=" + timestamp +
                ", isBadSample=" + isBadSample +
                '}';
    }
}

POJO 设计要点： 所有字段使用 public 修饰符，可让 Flink 在序列化时直接访问字段，避免反射开销。同时保留无参构造器，这是 Flink 类型系统识别 POJO 的必要条件。

3.2 模拟数据生成器设计

模拟数据生成器需要满足以下要求：

支持配置每秒生成的事件数量（QPS）
能够模拟正常点击与坏样本点击（如快速连续点击、异常时间戳、重复事件等）
事件时间戳应模拟真实场景，带有一定的随机延迟
支持多用户、多商品、多页面分布

我们设计一个 ClickEventGenerator 类，内部维护用户池、商品池、页面池，并通过随机概率控制坏样本的生成比例。

// ClickEventGenerator.java
import java.util.Random;
import java.util.UUID;

public class ClickEventGenerator {
    private static final String[] USERS = {"user_1001","user_1002","user_1003","user_1004","user_1005"};
    private static final String[] PAGES = {"page_home","page_detail","page_search","page_cart","page_order"};
    private static final String[] ITEMS = {"item_001","item_002","item_003","item_004","item_005"};
    private static final String[] CLICK_TYPES = {"view","add_cart","purchase","favorite","share"};

    private final Random random = new Random();
    private final double badSampleRatio; // 坏样本比例，例如 0.1 表示 10%

    public ClickEventGenerator(double badSampleRatio) {
        this.badSampleRatio = badSampleRatio;
    }

    public ClickEvent generate(long baseTimestamp) {
        String userId = USERS[random.nextInt(USERS.length)];
        String eventId = "evt_" + UUID.randomUUID().toString().substring(0,6);
        String pageId = PAGES[random.nextInt(PAGES.length)];
        String itemId = ITEMS[random.nextInt(ITEMS.length)];
        String clickType = CLICK_TYPES[random.nextInt(CLICK_TYPES.length)];
        long timestamp = baseTimestamp + random.nextInt(100); // 0~99ms 偏移
        boolean isBadSample = random.nextDouble() < badSampleRatio;

        // 如果是坏样本，模拟异常行为：时间戳异常或快速重复
        if (isBadSample) {
            // 模拟坏样本：时间戳跳变或重复事件
            if (random.nextBoolean()) {
                timestamp = baseTimestamp - 5000; // 时间戳回退 5 秒
            } else {
                // 保持正常时间戳，但标记为坏样本（后续逻辑会识别）
            }
        }

        return new ClickEvent(userId, eventId, pageId, itemId, clickType, timestamp, isBadSample);
    }
}

3.3 使用 DataStreamSource 接入流

Flink 提供了 SourceFunction 接口，允许我们自定义数据源。我们将实现一个 SimulatedClickSource，在 run() 方法中循环生成事件，并通过 ctx.collect() 发送到下游算子。

// SimulatedClickSource.java
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.util.concurrent.TimeUnit;

public class SimulatedClickSource implements SourceFunction<ClickEvent> {
    private volatile boolean running = true;
    private final int eventsPerSecond;
    private final double badSampleRatio;

    public SimulatedClickSource(int eventsPerSecond, double badSampleRatio) {
        this.eventsPerSecond = eventsPerSecond;
        this.badSampleRatio = badSampleRatio;
    }

    @Override
    public void run(SourceContext<ClickEvent> ctx) throws Exception {
        ClickEventGenerator generator = new ClickEventGenerator(badSampleRatio);
        long baseTimestamp = System.currentTimeMillis();
        long intervalMs = 1000 / eventsPerSecond;

        while (running) {
            long currentBase = System.currentTimeMillis();
            ClickEvent event = generator.generate(currentBase);
            ctx.collect(event);
            Thread.sleep(intervalMs);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}

关键设计决策： 使用 volatile boolean running 确保多线程环境下取消信号可见。在 run() 方法中通过 Thread.sleep() 控制发射速率，模拟真实场景下的均匀流量。

3.4 数据流拓扑与执行

在 Flink 主程序中，我们将 SimulatedClickSource 注册为数据源，并设置并行度。以下为完整的执行入口：

// ClickStreamJob.java
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;

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

        // 创建模拟数据源：每秒 1000 条，10% 坏样本
        SimulatedClickSource source = new SimulatedClickSource(1000, 0.1);
        DataStream<ClickEvent> clickStream = env.addSource(source)
                .name("simulated-click-source")
                .setParallelism(1); // 数据源通常并行度为 1，保证顺序

        // 打印到控制台验证（后续章节将替换为实际处理逻辑）
        clickStream.print();

        env.execute("Real-time Bad Sample Attribution Analysis");
    }
}

3.5 数据验证与调试

启动 Flink 作业后，控制台将输出类似以下格式的事件记录。通过观察 isBadSample 字段和时间戳，可以验证模拟数据是否符合预期。

ClickEvent{userId='user_1003', eventId='evt_a1b2c3', pageId='page_detail', itemId='item_002', clickType='purchase', timestamp=1710000000123, isBadSample=false}
ClickEvent{userId='user_1001', eventId='evt_d4e5f6', pageId='page_home', itemId='item_005', clickType='view', timestamp=1710000000087, isBadSample=true}
ClickEvent{userId='user_1005', eventId='evt_g7h8i9', pageId='page_search', itemId='item_001', clickType='add_cart', timestamp=1709999995123, isBadSample=true}

调试注意事项： 如果发现所有事件的时间戳完全相同，请检查 generate() 方法中是否正确使用了 baseTimestamp + random.nextInt(100) 的偏移逻辑。另外，坏样本的时间戳回退值（-5000ms）应明显偏离正常范围，以便后续窗口算子能够识别。

3.6 本章小结

通过本章的学习，我们完成了以下关键工作：

定义了 ClickEvent POJO，包含 7 个核心字段，满足 Flink 序列化要求
实现了 ClickEventGenerator，支持多用户、多商品、多页面分布，并可按比例注入坏样本
编写了 SimulatedClickSource，通过 SourceFunction 接口将模拟数据接入 Flink 流处理环境
配置了可调节的参数（QPS、坏样本比例），为后续压力测试和归因分析提供了灵活的数据源