在将输入解析为 JSON 并保留时，仅保留第一个 JSON 对象答案

【问题标题】：While parsing input as JSON and persisting, only firsts JSON object is persisted在将输入解析为 JSON 并保留时，仅保留第一个 JSON 对象
【发布时间】：2016-03-10 01:59:47
【问题描述】：

我正在尝试读取发布到我的管道订阅的主题的 JSON 数组，并将其保存到 BigQuery。我这样做时面临的问题是它只保留第一个对象，有人可以告诉我我在这里做错了什么。

    /** A DoFn that converts a table row from JSON into a BigQuery table row. */
  static class FormatAsTableRowFn extends DoFn<TableRow, TableRow> {
    private static final long serialVersionUID = 0;

    static TableSchema getSchema() {
        return new TableSchema().setFields(new ArrayList<TableFieldSchema>() {
              // Compose the list of TableFieldSchema from tableSchema.
              {
                add(new TableFieldSchema().setName("PillBoxID").setType("STRING").setMode("NULLABLE"));
                add(new TableFieldSchema().setName("Period").setType("STRING").setMode("NULLABLE"));
                add(new TableFieldSchema().setName("Time").setType("TIMESTAMP").setMode("NULLABLE"));
                add(new TableFieldSchema().setName("IsTaken").setType("STRING").setMode("NULLABLE"));
              }
        });
      }

    @Override
    public void processElement(ProcessContext c) {
        TableRow jsonRow = c.element();
        // Setup a date formatter to parse the date appropriately
          SimpleDateFormat ft = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
        try {

            TableRow bQueryRow = new TableRow()
             .set("PillBoxID", (String) jsonRow.get("PillBoxID"))
             .set("Period", (String) jsonRow.get("Period"))
             .set("Time",ft.format(ft.parse((String) jsonRow.get("Time"))))
             .set("IsTaken", (String) jsonRow.get("IsTaken"));
             LOG.error("Inside try" + bQueryRow.getF()); 
          c.output(bQueryRow);

       } catch (ParseException pe) {
           LOG.error("ParseException");     
           LOG.error(pe.getMessage());
       }
    }
}

而我的流水线代码如下所示，

    bigQueryPipeLine
    .apply(PubsubIO.Read.topic(options.getPubsubTopic()).withCoder(TableRowJsonCoder.of()))
    .apply(ParDo.of(new FormatAsTableRowFn()))
    .apply(BigQueryIO.Write.to(tableSpec)
        .withSchema(FormatAsTableRowFn.getSchema()));

【问题讨论】：

该代码看起来很合理。您是否查看过 Developer Console 中的 Dataflow 监控 UI？您可以在此处查看管道中的每个转换已处理了多少数据。还有一个指向 Cloud Logging 的链接，您可以在其中查看是否收到 ParseExceptions

标签： google-cloud-dataflow

【解决方案1】：

如果将输入 JSON 格式化为包含项目数组，则可以处理多条记录。

示例输入：

{
    "items":
    [
        {"PillBoxID":"ID5", "Period":"Morning", "Time":"2016-03-14T11:11:11", "IsTaken":"true"},
        {"PillBoxID":"ID6", "Period":"Afternoon", "Time":"2016-03-14T15:11:11", "IsTaken":"false"}
    ]
}

粗略的示例 processElement() 代码将这 2 个项目添加到 c.output() 以便稍后存储在 BigQuery 中。

    @Override
    public void processElement(ProcessContext c) throws DatastoreException, IOException{
        TableRow jsonRowObj = c.element();
        LOG.info("Original input:" + c.element().toPrettyString());
        ArrayList<Map> jsonRows = (ArrayList<Map>)jsonRowObj.get("items");
        Iterator<Map> iterator = jsonRows.iterator();

        while(iterator.hasNext()) {

            Map jsonRow =  (Map)iterator.next();
            // Setup a date formatter to parse the date appropriately
            SimpleDateFormat ft = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
            try {
                LOG.info("child input JSON: "+jsonRow.toString());
                TableRow bQueryRow = new TableRow()
                 .set("PillboxID", (String) jsonRow.get("PillBoxID"))
                 .set("Period", (String) jsonRow.get("Period"))
                 .set("Time",ft.format(ft.parse((String) jsonRow.get("Time"))))
                 .set("IsTaken", (boolean) Boolean.parseBoolean((String)jsonRow.get("IsTaken")));

              c.output(bQueryRow);

           } catch (ParseException pe) {
               LOG.error("");   
               LOG.error("ParseException " +pe.getMessage());

        }
    }
}

【讨论】：