调度程序不更新包文件答案

【问题标题】：Scheduler not updating package files调度程序不更新包文件
【发布时间】：2018-06-28 10:27:38
【问题描述】：

我正在 Cloud Composer 上开发 DAG；我的代码被分成一个主要的 python 文件和一个带有子文件夹的包，它看起来像这样：

my_dag1.py
package1/__init__.py
package1/functions.py
package1/package2/__init__.py
package1/package2/more_functions.py

我更新了package1/functions.py 上的一个函数以获取附加参数（并更新my_dag1.py 中的引用）。该代码将在我的本地环境中正确运行，并且在运行时我没有收到任何错误

gcloud beta composer environments run my-airflow-environment list_dags --location europe-west1

但是 Web UI 引发了 python 错误

TypeError: my_function() 得到了一个意外的关键字参数 '新参数'

只有在我重命名包文件夹后，错误才消失。

我怀疑这个问题与调度程序接收 my_dag1.py 而不是 package1/functions.py 有关。该错误突然出现，因为我在前几周进行了类似的更新。

知道如何在不重构整个代码结构的情况下解决此问题吗？

EDIT-1

Here's the link 到 Google 网上论坛的相关讨论

【问题讨论】：

能否分享您从相关文件中导入的内容？
导入使用以下语法：from package1.functions import my_function，它们在我的本地 Airflow 环境中正常工作（以及在我将属性添加到函数之前在 Cloud Composer 中）
我在 Cloud Composer 上遇到了一些类似的问题（但在本地环境中没有），调试起来真的很痛苦。您是否尝试删除您的文件 (gcloud beta composer environments storage dags delete --environment my-airflow-environment --location europe-west1 <your directory>) 并重新上传它们？这也可能是 Web UI 上的问题，但无法在 Cloud Composer 上重新启动 Web 服务器。作为最后的手段，您能否尝试在新创建的 Cloud Composer 环境中上传文件，看看是否可行？这可能有助于隔离您的问题。
我使用 Cloud Storage Web 界面删除了所有文件，但没有帮助。下次遇到此问题时，我将尝试使用 composer 控制台命令，看看是否有什么不同。感谢您的建议！

【解决方案1】：

我遇到了类似的问题。 “Broken DAG”错误不会在 Web UI 中消失。我猜这是 AirFlow 的 Web 服务器中的缓存错误。

背景。
1. 我创建了一个具有Airflow Plugin 功能的自定义运算符。
2. 导入自定义算子后，airflow Web UI一直显示Broken DAG错误，提示找不到自定义算子。
为什么我认为这是 Web 服务器 Airflow 中的错误？
1. 我可以使用命令airflow test手动运行DAG，所以导入应该是正确的。
2. 即使我从气流的/dags/文件夹中删除了相关的DAG文件，错误仍然存在。
这是我为解决此问题所做的工作。
1. 重新启动气流网络服务。（有时您只能通过此方法解决问题）。
2. 确保没有 DAG 正在运行，重新启动气流调度程序服务。
3. 确保没有 DAG 正在运行，重新启动气流工作器

希望它可以帮助遇到同样问题的人。

【讨论】：

【解决方案2】：

尝试重启网络服务器：

gcloud beta composer environments restart-web-server ENVIRONMENT_NAME --location=LOCATION

【讨论】：