如果要在java中使用apache beam，请首先在pom.xml文件中添加以下两个依赖项：

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-core</artifactId>
    <version>2.25.0</version>
</dependency>

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-runners-direct-java</artifactId>
    <version>2.25.0</version>
</dependency>

然后，为了达到你想要的输出，你可以这样做（这不是完美的，但给你的想法）

PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);
p
        .apply(TextIO.read().from("input.txt"))
        .apply(MapElements.into(TypeDescriptors.lists(TypeDescriptors.strings()))
                .via(l -> Arrays.asList(l.split(","))))
        .apply(Filter.by(element -> element.get(3).equals("Accounts")))
        .apply(MapElements.into(TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptors.integers()))
                .via(element -> KV.of(element.get(1), 1)))
        .apply(Combine.perKey((a, b) -> a + b))
        .apply(MapElements.into(TypeDescriptors.strings()).via(element -> element.getKey() + "=" + element.getValue()))
        .apply(TextIO.write().to("data/output").withNumShards(1).withSuffix(".txt"));

p.run().waitUntilFinish();

结果是：

Kyle=2
Kumiko=1