使用Apache Beam将重复字符串写入BigQuery

Question

我有一个包含字符串的数据流，看起来像JSONArrays。我想解析这些字符串并使用Apache Beam写入BigQuery表，但在写入重复字符串时出现错误。

以下是如何将字符串转换为tableRow:

    String dataString = "[{\"EMAIL\": [\"[e-mail]\"]}]";

    JSONArray jsonArray = new JSONArray(dataString);
    TableRow tableRow = new TableRow();

    for (int i = 0; i < jsonArray.length(); i++) {
      JSONArray emailArray = new JSONArray(jsonArray.getJSONObject(i).get("EMAIL").toString());

      tableRow.set("EMAIL", emailArray); //Results in error
    }

下面是我的BigQuery模式的样子：

[
  {
    "name": "EMAIL",
    "type": "STRING",
    "mode": "REPEATED"
  }
]

我已经设法使用Python为BigQuery表编写了一个类似的重复字符串，但无法使用Apache Beam来完成。我想我没有在tableRow中保存正确的键值对。我现在得到的错误是：

java.io.IOException: Insert failed: [{"errors":[{"debugInfo":"","location":"email","message":"This field is not a record.","reason":"invalid"}],"index":0}]

我需要关于如何保存一个类似的重复字符串到BigQuery而不创建一个记录的帮助，并将感谢任何意见或建议。提前道谢。

Answer 1

看来你想创造

1. 带有电子邮件地址串联字符串的一行，或
2. 每封电子邮件一行，或
3. 具有重复字段的一行。

注意，您的validfrom字段似乎属于string类型，而不是重复字段，除非它包装在分层架构中的重复字段中。

在您提供的示例代码中，您正在创建一个JSONARRAY并将其放入String字段，我认为这会导致一些问题，因为类型不兼容。如果希望将其保留为纯字符串字段，可以使用下面的解决方案1。

还要确保BigQuery中的列名与代码中的列名相匹配，我看到您同时使用了validfrom和email（但在发布的代码中可能是一个错误）。

如果您想在BigQuery中添加带有级联string字段的一行，可以使用以下方法：

// Initialize your final row
TableRow tableRow = new TableRow();

// Find email addresses
String [] emails = ... // your extraction logic

// Build a concatenated string of emails
String allEmails = String.join(";", emails);

// Add the string field to the row
tableRow.set('EMAILS', allEmails);

如果您想插入多行，请创建多个表行：

// Find email addresses
String [] emails = ... // your extraction logic

// Build a row per email
for(String email: emails) {
    // Initialize your final row
    TableRow tableRow = new TableRow();
    tableRow.set('EMAIL', email);
    
    // TODO: do something with the row (add to list, or ...)
}

如果要在BigQuery中添加带有重复字符串字段的一行，可以使用以下方法：

// Initialize your final row
TableRow tableRow = new TableRow();

// Find email addresses
String [] emails = ... // your extraction logic

// Build the repeated field
List<String> emailCells = new ArrayList<>();
for(String email: emails) {
    emailCells.add(email);
}

// Add the repeated field to the row
tableRow.set('EMAILS', emailCells);

如果这不是你的目标，请提供一些更多的细节。