如何在Apache Beam Java中将PCollection转换为List集合？

在Apache Beam中，PCollection是分布式数据的逻辑抽象，本身不存储数据，因此没有直接的get()或类似方法来提取数据。要将处理后的PCollection转换为List，需要结合管道运行和视图转换来实现，分两种常见场景：

一、本地调试/开发场景（DirectRunner） #

如果是在本地用DirectRunner运行管道，可以通过View.asIterable()将PCollection转换为可迭代视图，再在管道运行完成后提取数据：

修改后的完整代码 #

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.MapCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.runners.DirectRunner;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.View;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import com.google.common.collect.Lists;

public class BeamExample {
    public static void main(String[] args) {
        PipelineOptions options = PipelineOptionsFactory.create();
        // 指定用DirectRunner本地运行，确保能收集到本地结果
        options.setRunner(DirectRunner.class);
        Pipeline pipeline = Pipeline.create(options);

        List<Map<String, String>> inputDataList = ...; // 你的输入数据

        // 创建类型明确的PCollection（建议指定泛型，避免原始类型）
        PCollection<Map<String, String>> pcollection = pipeline.apply(
                Create.of(inputDataList)
                        .withCoder(MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()))
        );

        // 应用你的数据转换逻辑
        pcollection = pcollection.apply(...); // 替换为实际的转换步骤

        // 将PCollection转换为可迭代的视图，用于后续提取数据
        PCollectionView<Iterable<Map<String, String>>> outputView = pcollection.apply(View.asIterable());

        // 运行管道并等待完成
        PipelineResult result = pipeline.run();
        result.waitUntilFinish();

        // 从视图中提取数据并转换为List
        List<Map<String, String>> outputDataList = Lists.newArrayList(
                result.getPipelineResultInternal()
                        .getMaterialization(outputView)
        );

        // 现在可以使用outputDataList了
        System.out.println(outputDataList);
    }
}

二、测试场景（使用TestPipeline） #

如果是编写单元测试，可以用TestPipeline结合PAssert来获取实际结果：

import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.values.PCollection;
import com.google.common.collect.Lists;

public class BeamTest {
    @Test
    public void testPipeline() {
        TestPipeline pipeline = TestPipeline.create();

        List<Map<String, String>> inputDataList = ...; // 测试输入

        PCollection<Map<String, String>> pcollection = pipeline.apply(Create.of(inputDataList));
        // 应用转换逻辑
        PCollection<Map<String, String>> output = pcollection.apply(...);

        // 获取PCollection的实际结果并转为List
        Iterable<Map<String, String>> outputIterable = PAssert.that(output).getActual();
        List<Map<String, String>> outputDataList = Lists.newArrayList(outputIterable);

        // 进行断言或其他测试操作
    }
}

关键说明 #

1. Beam的延迟执行模型：管道定义阶段只是构建处理逻辑，必须调用run()才会触发实际的数据处理。
2. 分布式场景限制：如果是在分布式Runner（如DataflowRunner）上运行，不能直接将PCollection转为本地List——因为数据分布在多个节点，这种场景下通常是将结果写入外部存储（如数据库、GCS、HDFS等），而不是收集到本地List。

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