无法查明 GCP Composer (Airflow) DAG 任务失败的问题

Question

我不熟悉使用 Apache Airflow。 我的 dag 的一些操作员状态为失败。我试图了解错误的根源。

以下是问题的详细信息： 我的 dag 很大，它的某些部分是由子 dag 组成的。 我在 Composer UI 中注意到的是，失败的 Subdags 都在名为 download_file 的 task_id 中完成，它使用 XCom 和 GoogleCloudStorageDownloadOperator。

>> GoogleCloudStorageDownloadOperator(
    task_id='download_file',
    bucket="sftp_sef",
    object="{{task_instance.xcom_pull(task_ids='find_file') | first }}",
    filename="/home/airflow/gcs/data/zips/{{{{ds_nodash}}}}_{0}.zip".format(table)
)

上述 Subdag 中的日志没有显示任何有用的信息。

日志：

[2020-04-07 15:19:25,618] {models.py:1359} INFO - 依赖关系全部满足 对于 [2020-04-07 15:19:25,660] {models.py:1359} INFO - [2020-04-07 15:19:25,660] 的依赖项全部满足 {models.py:1577} 信息 -

----------------------------------------------- -------------------------------- 开始尝试第 10 次，共 1 次

[2020-04-07 15:19:25,685] {models.py:1599} 信息 - 正在执行 在 2020-04-06T11:44:31+00:00 [2020-04-07 15:19:25,685] {base_task_runner.py:118} 信息 - 运行：['bash'，'-c'，'气流运行 datamart_integration.consentement_email download_file 2020-04-06T11:44:31+00:00 --job_id 156313 --pool 集成 --raw -sd DAGS_FOLDER/datamart/datamart_integration.py --cfg_path /tmp/tmpacazgnve']

我不确定是否有我没有检查的地方...这是我的问题：

1. 我一般如何调试 Composer DAG 中的错误
2. 创建一个局部气流环境来运行& 在本地调试我的 dags？
3. 如何验证 XCom 中是否存在错误？

Answer 1

关于你的三个问题：

首先，在使用 Cloud Composer 时，您可以通过多种方式调试代码中的错误。根据documentation，您应该：

1. 检查气流日志。

这些日志与单个 DAG 任务相关。可以在 Cloud Storage 的日志文件夹和 Web Airflow 界面中查看它们。

当您创建 Cloud Composer 环境时，还会创建一个 Cloud Storage Bucket 并与之关联。因此，Cloud Composer 将单个 DAG 任务的日志存储在此存储桶内的日志文件夹中，每个工作流文件夹都有一个用于存放其 DAG 和子 DAG 的文件夹。你可以查看它的结构here。

关于 Airflow 网页界面，每 60 秒刷新一次。另外，您可以查看更多关于它的信息here。

1. 查看 Google Cloud 的操作套件。

您可以将 Cloud Monitoring 和 Cloud Logging 与 Cloud Composer 一起使用。 Cloud Monitoring 提供对云驱动应用的性能和整体运行状况的可见性，而 Cloud Logging 显示调度程序和工作容器生成的日志。因此，您可以根据需要同时使用两者或仅使用您认为更有用的一种。

1. 在 Cloud Console 中，检查页面上运行您的环境的 Google Cloud 组件的错误。

2. 在 Airflow Web 界面中，在 DAG 的 Graph View 中检查失败的任务实例。

因此，这些是在对 DAG 进行故障排除时推荐的步骤。

第二，关于测试和调试，建议您将生产环境和测试环境分开，避免DAG干扰。

此外，可以在本地测试您的 DAG，文档中有关于此主题的教程，here。本地测试允许您识别语法和任务错误。但是，我必须指出，无法检查/评估依赖关系以及与数据库的通信。

第三，一般来说，为了验证 Xcom 中的错误，您应该检查：

• 如果有任何错误代码/编号；
• 检查文档中的示例代码是否正确；
• 检查包是否被弃用；

我想指出，根据this documentation，GoogleCloudStorageDownloadOperator 的路径已更新为 GCSToLocalOperator。

此外，我还鼓励你看看这个：code 和 documentation 来检查 Xcom 语法和错误。

如果您需要进一步的帮助，请随时与我分享错误代码。