Java基于Apache Beam+Dataflow实现未知Schema JSONL转CSV通用管道

前置需求对齐 #

先明确要满足的核心要求：

• 输入为GCS存储的大体积JSONL文件，文件内JSON结构完全未知，管道开发阶段无法预定义固定Schema
• 适配多客户端异构字段场景：不同客户端上传的JSONL字段集完全无统一规范（比如客户端1字段为city、pincode，客户端2字段为Relation、Code，客户端3字段为Name、Session、Score、Completed），无需针对单个客户端写适配逻辑
• 自动提取单文件内所有JSON的全量键作为CSV表头，逐行映射字段值输出，输出CSV存回原JSONL所在的GCS存储桶
• 转换完成后支持状态标记，可通过REST API或事件机制设置完成标识（如transformation_done=true）

核心实现思路 #

Beam是分布式计算框架，不要做跨文件的Schema全局合并——单个JSONL文件本身对应单个客户端的固定结构，以单个文件为最小处理单元就能完全规避异构Schema冲突的问题，同时避免不必要的全局性能损耗：

1. 首先匹配GCS路径下所有待处理JSONL文件，每个文件独立走完整转换流程
2. 对单个文件先做全量键收集：遍历文件内所有JSON行，提取所有出现过的字段，对字段做字典序排序生成固定顺序的CSV表头，保证同个文件输出的表头顺序统一
3. 再次遍历文件内的JSON行，按照已确定的表头顺序逐行提取对应字段值，缺失字段填空字符串，按照CSV标准规范做特殊字符转义
4. 把生成的CSV内容写入原文件所在GCS桶，文件名和原JSONL保持一致、后缀替换为.csv
5. 单文件转换完成后触发通知逻辑：可以给GCS对象设置自定义元数据打完成标，也可以调用业务侧REST回调接口、发布事件消息供下游消费

关键代码实现（Java） #

所有逻辑基于Beam Java SDK、Dataflow运行时实现，核心片段如下：

// 1. 读取匹配规则下的所有GCS JSONL文件
PCollection<FileIO.ReadableFile> inputFiles = pipeline.apply(
    "MatchSourceJSONL",
    FileIO.match().filepattern("gs://target-bucket/input-path/*.jsonl")
).apply("ReadMatchedFiles", FileIO.readMatches());

// 定义两个TupleTag用于后续关联
final TupleTag<Set<String>> schemaTag = new TupleTag<>(){};
final TupleTag<String> lineTag = new TupleTag<>(){};

// 2. 分支1：收集单个文件的全量字段集合
PCollection<KV<String, Set<String>>> perFileSchemas = inputFiles.apply(
    "CollectKeysForSingleFile",
    ParDo.of(new DoFn<FileIO.ReadableFile, KV<String, Set<String>>>() {
        // 实例化JSON解析器，线程安全
        private final ObjectMapper objectMapper = new ObjectMapper();
        @ProcessElement
        public void process(@Element FileIO.ReadableFile file, OutputReceiver<KV<String, Set<String>>> out) throws IOException {
            String fileResourceId = file.getMetadata().resourceId().toString();
            Set<String> allKeys = new HashSet<>();
            // 逐行读文件收集字段
            try (BufferedReader reader = new BufferedReader(new InputStreamReader(file.open(), StandardCharsets.UTF_8))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    if (line.isBlank()) continue;
                    JsonNode jsonNode = objectMapper.readTree(line);
                    jsonNode.fieldNames().forEachRemaining(allKeys::add);
                }
            }
            out.output(KV.of(fileResourceId, allKeys));
        }
    })
);

// 3. 分支2：读取单个文件的所有JSON行，以文件路径为key
PCollection<KV<String, String>> perFileLines = inputFiles.apply(
    "ReadLinesForSingleFile",
    ParDo.of(new DoFn<FileIO.ReadableFile, KV<String, String>>() {
        @ProcessElement
        public void process(@Element FileIO.ReadableFile file, OutputReceiver<KV<String, String>> out) throws IOException {
            String fileResourceId = file.getMetadata().resourceId().toString();
            try (BufferedReader reader = new BufferedReader(new InputStreamReader(file.open(), StandardCharsets.UTF_8))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    if (!line.isBlank()) out.output(KV.of(fileResourceId, line));
                }
            }
        }
    })
);

// 4. 关联同个文件的Schema和行数据，生成标准CSV内容
PCollection<KV<String, String>> generatedCsv = KeyedPCollectionTuple
    .of(schemaTag, perFileSchemas)
    .and(lineTag, perFileLines)
    .apply("JoinSchemaAndLinesByFile", CoGroupByKey.create())
    .apply("BuildCSVContent", ParDo.of(new DoFn<KV<String, CoGbkResult>, KV<String, String>>() {
        private final ObjectMapper objectMapper = new ObjectMapper();
        @ProcessElement
        public void process(@Element KV<String, CoGbkResult> element, OutputReceiver<KV<String, String>> out) throws IOException {
            String fileId = element.getKey();
            CoGbkResult groupedData = element.getValue();
            // 拿到排序后的固定表头
            List<String> csvHeaders = groupedData.getOnly(schemaTag).stream()
                .sorted().toList();
            StringBuilder csvBuffer = new StringBuilder();
            // 用标准CSV工具类处理转义，不要手动拼接
            try (CSVPrinter printer = new CSVPrinter(csvBuffer, CSVFormat.DEFAULT)) {
                printer.printRecord(csvHeaders);
                for (String jsonLine : groupedData.getAll(lineTag)) {
                    JsonNode node = objectMapper.readTree(jsonLine);
                    List<String> rowValues = csvHeaders.stream()
                        .map(field -> node.has(field) ? node.get(field).asText() : "")
                        .toList();
                    printer.printRecord(rowValues);
                }
            }
            out.output(KV.of(fileId, csvBuffer.toString()));
        }
    })
);

// 5. 将生成的CSV写回原GCS桶
generatedCsv.apply("WriteCSVToGCS", ParDo.of(new DoFn<KV<String, String>, String>() {
    @ProcessElement
    public void process(@Element KV<String, String> csvData, OutputReceiver<String> out) throws IOException {
        String sourceJsonlPath = csvData.getKey();
        String targetCsvPath = sourceJsonlPath.replace(".jsonl", ".csv");
        try (WriteChannel writeChannel = FileSystems.create(
            FileSystems.matchNewResource(targetCsvPath, false),
            "text/csv"
        )) {
            writeChannel.write(ByteBuffer.wrap(csvData.getValue().getBytes(StandardCharsets.UTF_8)));
        }
        out.output(targetCsvPath);
    }
}))
// 6. 转换完成后打状态标记
.apply("MarkCompleteStatus", ParDo.of(new DoFn<String, Void>() {
    private transient Storage gcsClient;
    @Setup
    public void init() {
        gcsClient = StorageOptions.getDefaultInstance().getService();
    }
    @ProcessElement
    public void process(@Element String csvPath) {
        // 方式1：给GCS对象设置元数据标记完成
        BlobId blobId = BlobId.fromGsUtilUri(csvPath);
        gcsClient.setBlobMetadata(blobId, Map.of("transformation_done", "true"));
        // 方式2：如果需要REST回调，直接在这里发起HTTP请求到业务接口即可
        // 方式3：也可以发布Pub/Sub事件通知下游系统
    }
}));

pipeline.run();

大文件场景优化点 #

• 单文件大小超过10G时，不要在单个DoFn内全量加载文件做键收集，可以把文件拆分为固定大小的分片，每个分片先收集分片内的字段集合，再做分片级别的字段归并，降低单Worker的内存压力
• 不要自己手写CSV转义逻辑，直接用成熟的CSV工具库处理逗号、引号、换行符等特殊场景，避免输出格式损坏
• 增加坏行容错：JSON解析失败的行可以单独写入原桶下的/error目录，不要因为单条坏数据导致整个转换任务失败
• 可以配合GCS对象创建事件触发管道运行，新文件上传后自动启动转换，不用定时全量扫描存储桶

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