如何在Apache Beam中将CSV数据读取并存入PCollection

PCollection类型指定方法 #

PCollection的类型需要和集合内存储的元素结构一一对应，根据CSV处理需求选对应类型即可：

• 仅做原始文本搬运、不需要解析字段时，直接指定为PCollection<String>，每个元素对应CSV文件的一行原始文本
• 需要按字段做计算、关联等处理时，提前定义和CSV字段匹配的结构：Python端可以用字典、dataclass，Java端可以用自定义POJO类，PCollection的泛型直接指定为你定义的结构类型即可，比如存储用户信息的话就用PCollection[User]（Python）/PCollection<User>（Java）

具体实现代码 #

Python SDK 实现 #

如果是普通无嵌套换行的标准CSV，可以直接用内置的ReadFromText读取后逐行解析：

import apache_beam as beam
import csv
from io import StringIO
from apache_beam.pvalue import PCollection

# 单条CSV行解析逻辑
def parse_csv_line(line: str):
    csv_reader = csv.reader(StringIO(line))
    for row in csv_reader:
        # 按你的CSV字段顺序做映射，也可以返回自定义dataclass实例
        return {
            "user_id": row[0],
            "user_name": row[1],
            "score": int(row[2])
        }

if __name__ == "__main__":
    with beam.Pipeline() as pipeline:
        # 读取指定路径下的CSV文件，skip_header_lines=1表示跳过第一行表头
        raw_csv_lines: PCollection[str] = pipeline | "ReadCSV" >> beam.io.ReadFromText(
            file_pattern="/your/storage/path/data.csv", # 替换为本地、HDFS、对象存储等实际存储路径
            skip_header_lines=1
        )
        # 解析后得到类型明确的结构化PCollection
        parsed_csv_data: PCollection[dict] = raw_csv_lines | "ParseRows" >> beam.Map(parse_csv_line)

        # 后续直接对parsed_csv_data做业务变换即可

Java SDK 实现 #

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
import java.io.StringReader;

public class CsvToPCollection {
    // 定义和CSV字段匹配的POJO
    public static class UserScore {
        private String userId;
        private String userName;
        private Integer score;

        public String getUserId() { return userId; }
        public void setUserId(String userId) { this.userId = userId; }
        public String getUserName() { return userName; }
        public void setUserName(String userName) { this.userName = userName; }
        public Integer getScore() { return score; }
        public void setScore(Integer score) { this.score = score; }
    }

    public static void main(String[] args) {
        Pipeline pipeline = Pipeline.create();
        // 读取原始CSV行
        PCollection<String> rawLines = pipeline.apply(TextIO.read()
                .from("/your/storage/path/data.csv") // 替换为实际存储路径
                .withSkipHeaderLines(1));

        // 解析为结构化PCollection
        PCollection<UserScore> parsedData = rawLines.apply(ParDo.of(new DoFn<String, UserScore>() {
            @ProcessElement
            public void processElement(@Element String line, OutputReceiver<UserScore> out) {
                try (CSVParser parser = CSVParser.parse(new StringReader(line), CSVFormat.DEFAULT)) {
                    for (CSVRecord record : parser) {
                        UserScore score = new UserScore();
                        score.setUserId(record.get(0));
                        score.setUserName(record.get(1));
                        score.setScore(Integer.parseInt(record.get(2)));
                        out.output(score);
                    }
                } catch (Exception e) {
                    // 按需添加脏数据过滤、告警逻辑
                }
            }
        }));

        pipeline.run().waitUntilFinish();
    }
}

注意事项 #

如果你的CSV存在字段内含换行、转义引号、特殊分隔符等复杂格式，不要手动按行解析，直接使用Beam内置的CSVIO组件读取，组件会自动处理格式兼容问题，同时根据CSV表头生成带Schema的PCollection，不需要手动指定类型和写解析逻辑。

内容的提问来源于stack exchange，提问作者innovation