确定导致 Google Dataflow 作业失败的特定输入数据

Posted 2023-03-25

【中文标题】确定导致 Google Dataflow 作业失败的特定输入数据

【英文标题】：Determining specific input data which causes a Google Dataflow job to fail

【发布时间】：2015-03-04 19:31:52

【问题描述】：

我在使用 Dataflow 解析文本文件然后将相关数据放入 BQ 时遇到问题。该问题似乎是由文本文件中的一行格式错误的输入引起的。我能够修复错误，但它让我思考：如果我有一行输入数据流，有什么办法可以找出这条线？这将使 Dataflow 调试的一部分变得更加容易，尤其是当您的输入文件有几十亿行并且您必须追踪导致问题的那一行时。

例如，假设我正在向 BigQuery 发布我认为是整数的数据。我可能会像这样创建我的架构：

List<TableFieldSchema> fields = new ArrayList<>();
    fields.add(newTableFieldSchema().setName("ItemNum").setType("INTEGER"));

我可能会使用此函数将输入数据映射到 BigQuery 架构：

    public void processElement(ProcessContext c) 

        TableRow row = new TableRow();
        row.set("ItemNum", c.element()); 
        c.output(row);
    

但是当 Dataflow 遇到我的格式错误的输入（不是整数）时，我会收到如下错误：

工作流程失败。原因：（30d455a6f7aaaaaa）：项目“项目名称”中的 BigQuery 作业“dataflow_job_3518531384490999999”完成错误：作业错误：无法将值转换为整数（错误值或超出范围）。，错误：无法转换值到整数（错误值或超出范围）。，错误：无法将值转换为整数（错误值或超出范围）。，错误：无法将值转换为整数（错误值或超出范围）。，错误：无法将值转换为整数（错误值或超出范围）。错误：无法将值转换为整数（错误值或超出范围）。

在这种特殊情况下，我应该在尝试将其放入 BigQuery 之前验证我的输入是否为预期的整数（然后记录任何未通过验证的数据）。但是一般问题仍然存在——假设我想查看导致此错误的输入，因为（我认为）我已经执行了所有适当的输入验证，并且不知道哪种格式错误的数据可能会导致此错误。我该怎么做？我在想某种 try/catch 类型的技巧（可能涉及日志消息）可能会起作用，但我不确定如何做到这一点。

谢谢！

【问题讨论】：

这可能不是您的问题，但我有一个 5GB 的 .csv 数据文件，其中某些记录在某些字段中有不必要的逗号和额外引号（例如，“Cleveland, OH”而不是“Cleveland OH” ）。这些导致 Big Query 无法加载数据集。在将数据加载到谷歌云存储和 BQ 之前，我使用 python 脚本在本地重写了 .csv 文件以删除无关的标点符号。之后一切正常。

我的问题类似。我没有对传入的数据进行适当的验证检查，并且输入格式错误。当遇到意外的输入时，我正在做一些类型转换。从来没有发现确切的错误，只知道在类型转换中添加额外的逻辑来处理意外输入似乎可以解决问题。

【参考方案1】：

您建议的方法（使用 try/catch，分别记录您的解析错误）是目前的好方法。我们正在积极研究使管道编写者能够处理此类问题的选项。

【讨论】：

很高兴听到。我期待着尝试你们想出的任何东西！