Apache Beam-Reading JSON和Stream

Question

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

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

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

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