【发布时间】：2018-02-05 08:25:06
【问题描述】：
在我的管道中，FileIO.readMatches() 转换读取带有有效 JSON 数组的大 JSON 文件（大约 300-400MB）并将FileIO.ReadableFile 对象返回到下一个转换。我的任务是从该 JSON 数组中读取每个 JSON 对象，添加新属性并输出到下一个转换。
目前我解析 JSON 文件的代码如下所示：
// file is a FileIO.ReadableFile object
InputStream bis = new ByteArrayInputStream(file.readFullyAsBytes());
// Im using gson library to parse JSON
JsonReader reader = new JsonReader(new InputStreamReader(bis, "UTF-8"));
JsonParser jsonParser = new JsonParser();
reader.beginArray();
while (reader.hasNext()) {
JsonObject jsonObject = jsonParser.parse(reader).getAsJsonObject();
jsonObject.addProperty("Somename", "Somedata");
// processContext is a ProcessContext object
processContext.output(jsonObject.toString());
}
reader.close();
在这种情况下，文件的全部内容将在我的记忆中，这带来了获取 java.lang.OutOfMemoryError 的选项。我正在寻找解决方案来一一读取所有 JSON 对象，而无需将整个文件保存在我的内存中。
可能的解决方案是使用来自对象FileIO.ReadableFile 的方法open()，它返回ReadableByteChannel 通道，但我不确定如何使用该通道专门从该通道读取一个JSON 对象。
更新的解决方案 这是我更新的解决方案，逐行读取文件
ReadableByteChannel readableByteChannel = null;
InputStream inputStream = null;
BufferedReader bufferedReader = null;
try {
// file is a FileIO.ReadableFile
readableByteChannel = file.open();
inputStream = Channels.newInputStream(readableByteChannel);
bufferedReader = new BufferedReader(new InputStreamReader(inputStream, "UTF-8"));
String line;
while ((line = bufferedReader.readLine()) != null) {
if (line.length() > 1) {
// my final output should contain both filename and line
processContext.output(fileName + file);
}
}
} catch (IOException ex) {
logger.error("Exception during reading the file: {}", ex);
} finally {
IOUtils.closeQuietly(bufferedReader);
IOUtils.closeQuietly(inputStream);
}
我发现此解决方案不适用于在 n1-standard-1 机器上运行的 Dataflow，并引发 java.lang.OutOfMemoryError: GC overhead limit exceeded 异常并在 n1-standard-2 机器上正常工作。
【问题讨论】：
-
整个 JSON 文件是否需要在一个转换步骤中进行解析？您可以使用
TextIO类来读取 JSON 文件并单独解析每个 JSON 数组，而不必将整个文件保存在内存中。您能否提供更多关于如何在代码示例中定义管道的上下文？ -
不幸的是，我不能在我的解决方案中使用 TextIO，因为我需要在同一个转换步骤中使用文件名和文件内容。我问了这个问题here。在我目前的帖子中，我没有提到那部分来简化我的问题。我的想法是获取 FileIO.ReadableFile 作为我的转换步骤的输入，从该输入对象中提取文件名和文件内容，从文件中解析每一行并将该文件与文件名中的附加信息一起输出到下一个转换步骤。
标签： java json google-cloud-dataflow apache-beam google-cloud-pubsub