java - Apache Beam - 读取 JSON 和 Stream

Question

我正在编写 Apache Beam 代码，我必须读取放置在项目文件夹中的 JSON 文件，然后读取数据并将其流式传输。

这是读取 JSON 的示例代码。这是正确的做法吗？

PipelineOptions options = PipelineOptionsFactory.create();
options.setRunner(SparkRunner.class);

Pipeline p = Pipeline.create(options);

PCollection<String> lines = p.apply("ReadMyFile", TextIO.read().from("/Users/xyz/eclipse-workspace/beam-prototype/test.json"));
System.out.println("lines: " + lines);

或者我应该使用，

p.apply(FileIO.match().filepattern("/Users/xyz/eclipse-workspace/beam-prototype/test.json"))

我只需要读取下面的 json 文件。从此文件中读取完整的testdata，然后将其流式传输。

{
“testdata":{
“siteOwner”:”xxx”,
“siteInfo”:{
“siteID”:”id_member",
"siteplatform”:”web”,
"siteType”:”soap”,
"siteURL”:”www”,
}
}
}

上面的代码不是读取json文件，而是像打印

lines: ReadMyFile/Read.out [PCollection]

，您能指导我提供示例引用吗？

Answer 1

This is the sample code to read JSON. Is this correct way of doing it?

快速回答您的问题，是的。您的示例代码是读取包含 JSON 的文件的正确方法，其中文件的每一行都包含一个 JSON 元素。 TextIO 输入转换逐行读取文件，因此如果单个 JSON 元素跨越多行，则它将无法解析。

第二个代码示例具有相同的效果。

The above code is not reading the json file, it is printing like

打印结果是预期的。变量 lines 实际上并不包含文件中的 JSON 字符串。 lines 是 String 的 PCollection；它只是表示应用变换后管道的状态。可以通过应用后续转换来访问管道中的元素。可以在转换的实现中访问实际的 JSON 字符串。