使用 DoFn 解析 Json 时的奇怪行为答案

【问题标题】：Weird behavior while parsing Json with a DoFn使用 DoFn 解析 Json 时的奇怪行为
【发布时间】：2018-02-09 17:18:06
【问题描述】：

我正在尝试解析一些我存储在谷歌云存储中的 Json。我正在使用 apache Beam 创建一个管道，该管道读取存储的 json，然后将其写入云 sql 数据库。我在编写解析方法时遇到了一些奇怪的行为。

这是我的 Json：

        [{
            "projectid": "Reminder101",
            "reminderkey": "001",
            "localid": "01",
            "timestamp": "2018-01-24 12:00"
        },
        {
            "projectid": "Reminder101",
            "reminderkey": "002",
            "localid": "02",
            "timestamp": "2018-01-25 9:00"
        },
        {
            "projectid": "Reminder101",
            "reminderkey": "003",
            "localid": "03",
            "timestamp": "2018-02-01 18:00"
        },
        {
            "projectid": "Reminder101",
            "reminderkey": "004",
            "localid": "04",
            "timestamp": "2018-02-6 15:35"
        },
        {
            "projectid": "USReminder101",
            "reminderkey": "001",
            "localid": "01",
            "timestamp": "2018/01/30 21:00"
        }
    ]

这是我的 json 解析方法（JsonHolder 只是一个 pojo）：

static class ParseJsonDoFn extends DoFn<String, List<JsonHolder>> {
  @ProcessElement
  public void processElement(ProcessContext context) {
    String incomingInfo = context.element();
    Gson gson = new Gson();
    System.out.println(incomingInfo.toString());
    Type type = new TypeToken<List<JsonHolder>>(){}.getType();
    List<JsonHolder> jsonholders = gson.fromJson(incomingInfo, type);
    context.output(jsonholders);
  }
}

现在当我像这样运行我的管道时，我得到了错误：

Exception in thread "main" 
org.apache.beam.sdk.Pipeline$PipelineExecutionException: 
com.google.gson.JsonSyntaxException: 
com.google.gson.stream.MalformedJsonException: Expected value at line 1 
column 2 path $

System.out.println();显示：

"timestamp": "2018-02-01 18:00"
},
    "reminderkey": "004",
[{
    "localid": "01",
},
    "timestamp": "2018-01-24 12:00"
    "reminderkey": "002",

但是，如果我创建一个未格式化的单行 Json 文件，它解析得很好。我可以推断的是，Json 文件在每一行结束后都被拆分，但我终其一生都无法弄清楚为什么或如何更正它。

干杯。

【问题讨论】：

您使用哪种输入转换来读取 JSON 文件？我相信所有用于读取文件的转换都会逐行读取文件。因此，格式化的多行 JSON 对象将被分解。我认为没有办法解决这个问题。
@Andrew 谢谢你提供的信息！

标签： java json apache-beam

【解决方案1】：

TextIO.read() 读取文本文件，将每一行作为 PCollection 的单独元素返回。这样做是为了使其可以处理无限大小的文件，而无需将文件内容加载到内存中。

如果您的输入格式不是行分隔的，您可能需要更灵活的 FileIO：match() 来查找与您感兴趣的文件模式匹配的文件， read() 自动解压缩并方便地处理文件内容作为 ReadableFile。然后使用 DoFn 以任何你想要的方式解析文件。

【讨论】：

完美，感谢您提供的信息。我会试试你的建议，看看我能不能让它发挥作用！