Flink左外连接：使用CSV文件数据丰富流数据

阿华AIGC实验室

2026-5-20

针对你在Flink 1.4.0中需要通过左外连接用CSV数据丰富Employee流的需求，我来一步步拆解解决方案——因为是流处理场景，我们需要根据你的 enrichment 数据是静态还是动态来选择不同的实现方式：

情况1：Enrichment数据是动态更新的（或需要按窗口匹配）

如果你的enrichmentsStream是持续产生数据的（比如CSV是分批加载的，后续还有新的 enrichment 数据流入），那么基于窗口的左外连接是合适的方案。步骤如下：

按共享id字段对两个流做keyBy
首先需要将两个流都按照id进行分区，这样相同id的Employee和Enrichements会被分发到同一个并行任务中：

KeyedStream<Employee, String> keyedEmployees = employeeStream.keyBy(Employee::getId);
KeyedStream<Enrichements, String> keyedEnrichments = enrichmentsStream.keyBy(Enrichements::getId);

定义窗口
流处理中必须通过窗口来限定匹配的时间范围，比如我们选择5分钟的滚动窗口（你可以根据业务需求调整窗口类型和大小）：

WindowedStream<Employee, String, TimeWindow> employeeWindow = 
    keyedEmployees.window(TumblingProcessingTimeWindows.of(Time.minutes(5)));
WindowedStream<Enrichements, String, TimeWindow> enrichmentWindow = 
    keyedEnrichments.window(TumblingProcessingTimeWindows.of(Time.minutes(5)));

执行左外连接并处理结果
使用leftOuterJoin方法，指定匹配条件为id相等，然后在JoinFunction中处理匹配结果——左外连接的特点是即使没有匹配到对应的Enrichements，原Employee也会被输出：

DataStream<Employee> enrichedEmployeeStream = employeeWindow.leftOuterJoin(enrichmentWindow)
    .where(Employee::getId)
    .equalTo(Enrichements::getId)
    .apply(new JoinFunction<Employee, Enrichements, Employee>() {
        @Override
        public Employee join(Employee emp, Enrichements enrich) {
            // 左外连接中，enrich可能为null（没有匹配到对应id的数据）
            if (enrich != null) {
                // 在这里给Employee字段赋值，示例如下：
                emp.setDepartment(enrich.getDepartment());
                emp.setSalaryGrade(enrich.getSalaryGrade());
                // 其他需要丰富的字段同理
            }
            // 无论是否匹配到，都返回Employee（未匹配则保持原字段）
            return emp;
        }
    });

情况2：Enrichment数据是静态的（一次性加载的大型CSV）

如果你的enrichmentsStream是一次性加载的静态CSV数据，后续不会有更新，那么用RichMapFunction会更高效（因为Flink 1.4还不支持BroadcastProcessFunction），避免窗口带来的延迟：

实现RichMapFunction加载静态数据
在RichMapFunction的open方法中加载CSV数据到内存中的Map，然后在map方法中直接根据id匹配：

DataStream<Employee> enrichedEmployeeStream = employeeStream.map(new RichMapFunction<Employee, Employee>() {
    private Map<String, Enrichements> enrichmentCache;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 这里实现CSV加载逻辑，将数据存入Map（key为id，value为Enrichements）
        // 注意：如果CSV很大，确保文件存储在分布式文件系统（如HDFS），或每个TaskManager本地都有副本
        enrichmentCache = loadCsvIntoMap("/path/to/your/large-enrichment.csv");
    }

    @Override
    public Employee map(Employee emp) throws Exception {
        Enrichements enrich = enrichmentCache.get(emp.getId());
        if (enrich != null) {
            // 丰富Employee字段
            emp.setDepartment(enrich.getDepartment());
            emp.setBenefits(enrich.getBenefits());
        }
        return emp;
    }
});

// 辅助方法：加载CSV到Map，你需要根据CSV格式实现具体解析逻辑
private Map<String, Enrichements> loadCsvIntoMap(String filePath) throws IOException {
    Map<String, Enrichements> map = new HashMap<>();
    // 示例：用BufferedReader读取CSV并解析
    try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) {
        String line;
        // 跳过表头
        reader.readLine();
        while ((line = reader.readLine()) != null) {
            String[] parts = line.split(",");
            String id = parts[0];
            String department = parts[1];
            // 解析其他字段...
            Enrichements enrich = new Enrichements(id, department, ...);
            map.put(id, enrich);
        }
    }
    return map;
}