以下是“向现有 DataFrame 添加哈希”的任务如何从花费几天时间到消耗几乎整个冲刺的过程。
2022 年第二季度，我开始开发一个数据管道，该管道从 REST 服务获取市场数据并将其存储在 BigQuery 表中。这是管道的高级解释。有趣的部分是如何查询数据，将其转换为 DataFrame，然后使用 AirFlow 的 GCSToBigQueryOperator 上传 BigQuery 表。
最初，看起来写起来很简单，但 Airflow 的“幂等”原理给它增加了一些挑战。从该 REST 服务获取的内容由另一个表决定，即使 JOB 是幂等的，它用作参考的表也可能在两次运行之间发生变化。经过额外的时间，与数据工程师的沟通管道已于 2022 年第三季度末准备就绪。
快进到 2024 年第一季度。此时，我们有更多用户访问数据，并且我们意识到我们的查询模式没有正确使用分区。或者更确切地说，我们想要基于字符串列访问数据，但无法在 BigQuery 中对字符串列进行分区。这导致扫描大量数据并经常达到每日配额。
这促使我们考虑如何根据字符串列对数据进行分区。我们的数据工程师建议使用 FarmHash 并附加模运算将该字符串列转换为整数。在概念验证中，这减少了近 90% 的扫描，查询性能提高了 3-5 倍。我们决定继续将此作为最终解决方案。我们所需要的只是：
为了在 Python 中计算 FarmHash 指纹，有一个 pyfarmhash 模块。我安装了该模块并使用下面的代码来计算哈希值，并且在本地一切都按预期工作。
def get_hash(val: str) -> int:
return additonal_logic(pyfarmhash.fingerprint64(...))
df[‘hash’] = df[‘Col’].apply(get_hash)
随着所有测试的通过，现在是时候将代码推送到 Airflow 并运行它了。我没想到在这个阶段会出现任何问题。事实上，我很高兴一切都按计划并在预计的时间内进行。
怀着愉快的心情和充满信心，我推动了我的改变，开始了工作，然后等待了 10-15 分钟让它完成。与此同时，我切换到另一项任务。很快，我收到了一封来自 Airflow 的意外失败电子邮件。我查看了日志，惊讶地发现安装 pyfarmhash 模块时失败了！
为了帮助您理解问题，我需要解释一下作业的结构。该作业有以下步骤：
在此过程中，下载数据的task-1是一个单独的Python模块。为了运行它，我使用了 Airflow 中的 PythonVirtualenvOperator。该操作符允许您根据需要指定包，然后将它们安装在新创建的虚拟环境中。安装包后，它的所有依赖项也会安装，您就可以开始使用了。
我添加了 pyfarmhash 作为下载数据的模块的依赖项，其他一切保持不变。但它失败了！为什么？
pyfarmhash 是一个用 C/C++ 实现的哈希库。安装后，它需要 GCC 来编译软件包，而 Airflow 主机上不存在该软件包。在 Airflow 主机上不安装 GCC 是有道理的，但不幸的是，这对我来说是一个障碍。
我寻找了 pyfarmhash 包的纯 Python 实现，但没有。然后，我寻找车轮包装，但同样没有。我考虑过构建轮子包并推动它们，但这将导致在内部提供轮子包的长期责任。我想避免额外的、类似于解决方法的步骤。我探索了所有选项，并与维护 Airflow 的团队进行了讨论。他们建议创建一个 Docker 镜像并在 KubernetesPodOperator 中运行它。这是一个不错的选择，因为我可以控制环境并包含所需的任何内容，而无需依赖外部环境。此外，该解决方案没有解决方法。唯一的短期缺点是需要更多时间来实施。
在开始使用基于 Docker 的解决方案之前，我已经在这项任务上花费了大约 16-20 个小时。对于基于 Docker 的解决方案，我还需要：
由于我不再在 Airflow 中使用 PythonVirtualEnvOperator，我决定完全删除它并改进工作流程。我必须更改 python 包以具有开始下载和清除逻辑的入口点
我又花了 30-36 个小时才准备好了 Docker 镜像的最终解决方案，这需要 6-7 个工作日，加上最初的 2 天，它变成了一项漫长的冲刺任务。
我回顾这一点并想知道，我不得不放弃工作解决方案，更改模块结构，创建 docker 映像，更改 10 多个 AirFlow 作业以使用 Docker 映像执行任务，处理这个现实并克服最初的挫败感。所有这一切只是因为，“单个 python 模块需要“gcc”来编译！”
以上是估计编码任务：可能会出现什么问题？的详细内容。更多信息请关注PHP中文网其他相关文章！
