估计编码任务：可能会出现什么问题？

以下是“向现有 dataframe 添加哈希值”的任务如何从花费几天时间到消耗几乎整个冲刺的过程。

2022 年第二季度，我开始开发一个数据管道，该管道从 rest 服务获取市场数据并将其存储在 bigquery 表中。这是管道的高级解释。有趣的部分是如何查询数据，将其转换为 dataframe，然后使用 airflow 的gc++stobigqueryoperator上传 bigquery 表。

起初，写起来似乎很简单，但 airflow 的“幂等”原理给它增加了一些挑战。从该 rest 服务获取的内容由另一个表决定，即使 job 是幂等的，它用作参考的表也可能在两次运行之间发生变化。经过额外的时间，与数据工程师的沟通管道已于 2022 年第三季度末准备就绪。

快进到 2024 年第一季度。此时，我们有更多的用户访问数据，并且我们意识到我们的查询模式没有正确使用分区。或者更确切地说，我们想要基于字符串列访问数据，但无法在 bigquery 中对字符串列进行分区。这导致扫描大量数据并经常达到每日配额。

这促使我们考虑如何根据字符串列对数据进行分区。我们的数据工程师建议使用 farmhash 并附加模运算将该字符串列转换为整数。在概念验证中，这减少了近 90% 的扫描，查询性能提高了 3-5 倍。我们决定继续将此作为最终解决方案。我们所需要的只是：

创建带有farmhash指纹的表
更改管道来计算指纹
上传数据。

为了在python中计算farmhash指纹，有一个pyfarmhash模块。我安装了该模块并使用下面的代码来计算哈希值，并且在本地一切都按预期工作。

def get_hash(val: str) -&gt; int:
    return additonal_logic(pyfarmhash.fingerprint64(...))

df[‘hash’] = df[‘Col’].apply(get_hash)

登录后复制

随着所有测试的通过，现在是时候将代码推送到 airflow 并运行它了。我没想到在这个阶段会出现任何问题。事实上，我很高兴一切都按计划并在预计的时间内进行。

带着愉快的心情和充满信心，我推动了我的改变，开始了工作，然后等待了10-15分钟让它完成。与此同时，我切换到了另一项任务。很快，我收到了一封来自 airflow 的意外失败电子邮件。我查看了日志，惊讶地发现安装 pyfarmhash 模块时失败了！

为了帮助您理解问题，我需要解释一下作业的结构。该工作有以下步骤：

下载 parquet 格式的数据
上传到gcs存储桶
删除现有数据；如果有的话。（避免重复数据）
将数据上传到bq表。

在此过程中，下载数据的task-1是一个单独的python模块。为了运行它，我使用了 airflow 中的pythonvirtualenvoperator。该操作符允许您根据需要指定包，然后将它们安装在新创建的虚拟环境中。安装包后，它的所有依赖项也会安装，您就可以开始使用了。

我添加了pyfarmhash作为下载数据的模块的依赖项，其他一切保持不变。但它失败了！为什么？

pyfarmhash是一个用c/c++实现的哈希库。安装后，它需要 gcc 来编译软件包，而 airflow 主机上不存在该软件包。在 airflow 主机上不安装 gcc 是有道理的，但不幸的是，这对我来说是一个障碍。

我寻找了 pyfarmhash 包的纯 python 实现，但没有。然后，我寻找车轮包装，但同样没有。我考虑过构建轮子包并推动它们，但这将导致在内部提供轮子包的长期责任。我想避免额外的、类似于解决方法的步骤。我探索了所有选项，并与维护 airflow 的团队讨论了它们。他们建议创建一个 docker 镜像并在 kubernetespodoperator 中运行它。这是一个不错的选择，因为我可以控制环境并包含所需的任何内容，而无需依赖外部环境。此外，该解决方案没有解决方法。唯一的短期缺点是需要更多时间来实施。

在开始使用基于 docker 的解决方案之前，我已经在这项任务上花费了大约 16-20 个小时。对于基于 docker 的解决方案，我还需要：