根据从数据流加载Bigquery表时我们如何设置maximum_bad_records？目前没有办法在从数据流加载数据到BigQuery时设置maxBadRecords配置。建议是在将Dataflow作业中的行插入BigQuery之前验证它们。

如果我有TableSchema和TableRow，我如何确保该行可以安全地插入到表中？

一定有比遍历模式中的字段，查看它们的类型并查看行中值的类更简单的方法，对吗？这似乎容易出错，并且该方法必须是万无一失的，因为如果无法加载单行，整个管道都会失败。

更新：

我的用例是一个ETL作业，一开始它将运行在云存储上的JSON（每行一个对象）日志上，并批量写入BigQuery，但后来会从PubSub读取对象并连续写入BigQuery。这些对象包含许多在BigQuery中不必要的信息，还包含甚至不可能在模式中描述的部分（基本上是自由形式JSON有效负载）。像时间戳这样的东西也需要格式化以与BigQuery一起使用。这个作业会有一些变体在不同的输入上运行并写入不同的表。

从理论上讲，这不是一个非常困难的过程，它需要一个对象，提取一些属性（50-100），格式化其中一些并将对象输出到BigQuery。我或多或少只是遍历属性名称列表，从源对象中提取值，查看配置以查看属性是否应该以某种方式格式化，必要时应用格式化（这可能是降级，将毫秒时间戳除以1000，从URL中提取主机名等），并将值写入TableRow对象。

我的问题是数据很乱。有几亿个对象，有些看起来不像预期的那样，这很少见，但是有了这些卷，仍然会发生罕见的事情。有时应该包含字符串的属性包含一个整数，反之亦然。有时应该有字符串的数组或对象。

理想情况下，我想获取我的TableRow并将其传递给TableSchema并询问“这可以吗？”。

由于这是不可能的，我所做的是查看TableSchema对象并尝试自己验证/转换值。如果TableSchema说属性是STRING类型，我在将其添加到TableRow之前运行value. toString（）。如果它是INTEGER，我检查它是Integer，Long或BigInteger等等。这种方法的问题是我只是猜测BigQuery中的工作方式。FLOAT将接受哪些Java数据类型？对于TIMESTAMP？我认为我的验证/转换捕获了大多数问题，但总是有异常和边缘情况。

根据我非常有限的经验，如果单行未能通过BigQuery的验证（就像常规加载一样，除非maxBadRecords设置为足够大的数字），整个工作管道（作业？工作流？不确定正确的术语）都会失败。它也会因表面上有帮助的消息而失败，例如“BigQuery导入作业dataflow_job_xxx”失败。原因：（5db0b2cdab1557e0）：项目"xxx"中的BigQuery作业"dataflow_job_xxx"以错误结束： errorResult：为非记录字段指定的JSON映射，错误：为非记录字段指定的JSON映射，错误：为非记录字段指定的JSON映射，错误：为非记录字段指定的JSON映射，错误：为非记录字段指定的JSON映射，错误：为非记录字段指定的JSON映射。也许在某个地方可以看到更详细的错误消息，可以告诉我它是哪个属性以及值是什么？如果没有这些信息，它还不如说“坏数据”。

据我所知，至少在批处理模式下运行时，Dataflow会将TableRow对象写入Cloud Storage中的暂存区，然后一旦一切就绪，就开始加载。这意味着我无处捕获任何错误，当BigQuery加载时，我的代码不再运行。我还没有在流式模式下运行任何作业，但我不确定那里会有什么不同，从我（公认有限）的理解基本原理是一样的，只是批处理大小更小。

人们使用Dataflow和BigQuery，所以不可能不担心整个管道因为一个错误的输入而停止工作。人们是如何做到的？