【发布时间】：2019-08-07 20:54:34
【问题描述】：
我正在编写一个 Airflow DAG，但遇到了一些函数问题。我正在尝试通过将数据打印到标准输出并使用logging 库来进行调试。
我的 DAG 示例是：
from datetime import timedelta
import airflow
import logging
from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.contrib.hooks.datadog_hook import DatadogHook
def datadog_event(title, text, dag_id, task_id):
hook = DatadogHook()
tags = [
f'dag:{dag_id}',
f'task:{task_id}',
]
hook.post_event(title, text, tags)
def datadog_event_success(context):
dag_id = context['task_instance'].dag_id
task_id = context['task_instance'].task_id
text = f'Airflow DAG failure for {dag_id}\n\nDAG: {dag_id}\nTasks: {task_id}'
title = f'Airflow DAG success for {dag_id}'
logging.info(title)
logging.info(text)
logging.info(dag_id)
logging.info(task_id)
datadog_event(title, text, dag_id, task_id)
args = {
'owner': 'airflow',
'start_date': airflow.utils.dates.days_ago(2),
}
dag = DAG(
dag_id='example_callback',
default_args=args,
schedule_interval='*/5 * * * *',
dagrun_timeout=timedelta(minutes=60),
on_success_callback=datadog_event_success,
)
my_task = DummyOperator(
task_id='run_this_last',
dag=dag,
)
在运行过程中出现错误：
airflow[9490]: Process DagFileProcessor4195-Process:
airflow[9490]: Traceback (most recent call last):
airflow[9490]: File "/usr/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap
airflow[9490]: self.run()
airflow[9490]: File "/usr/lib/python3.6/multiprocessing/process.py", line 93, in run
airflow[9490]: self._target(*self._args, **self._kwargs)
airflow[9490]: File "/home/airflow/virtualenv/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 148, in _run_file_processor
airflow[9490]: result = scheduler_job.process_file(file_path, pickle_dags)
airflow[9490]: File "/home/airflow/virtualenv/lib/python3.6/site-packages/airflow/utils/db.py", line 74, in wrapper
airflow[9490]: return func(*args, **kwargs)
airflow[9490]: File "/home/airflow/virtualenv/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1542, in process_file
airflow[9490]: self._process_dags(dagbag, dags, ti_keys_to_schedule)
airflow[9490]: File "/home/airflow/virtualenv/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1239, in _process_dags
airflow[9490]: self._process_task_instances(dag, tis_out)
airflow[9490]: File "/home/airflow/virtualenv/lib/python3.6/site-packages/airflow/utils/db.py", line 74, in wrapper
airflow[9490]: return func(*args, **kwargs)
airflow[9490]: File "/home/airflow/virtualenv/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 732, in _process_task_instances
airflow[9490]: run.update_state(session=session)
airflow[9490]: File "/home/airflow/virtualenv/lib/python3.6/site-packages/airflow/utils/db.py", line 70, in wrapper
airflow[9490]: return func(*args, **kwargs)
airflow[9490]: File "/home/airflow/virtualenv/lib/python3.6/site-packages/airflow/models/dagrun.py", line 318, in update_state
airflow[9490]: dag.handle_callback(self, success=True, reason='success', session=session)
airflow[9490]: File "/home/airflow/virtualenv/lib/python3.6/site-packages/airflow/utils/db.py", line 70, in wrapper
airflow[9490]: return func(*args, **kwargs)
airflow[9490]: File "/home/airflow/virtualenv/lib/python3.6/site-packages/airflow/models/dag.py", line 620, in handle_callback
airflow[9490]: callback(context)
airflow[9490]: File "/home/airflow/analytics/etl_v2/airflow_data/dags/example_bash_operator_andy.py", line 68, in datadog_event_success
airflow[9490]: datadog_event(title, text, dag_id, task_id)
airflow[9490]: File "/home/airflow/analytics/etl_v2/airflow_data/dags/example_bash_operator_andy.py", line 45, in datadog_event
airflow[9490]: hook.post_event(title, text, tags)
airflow[9490]: File "/home/airflow/virtualenv/lib/python3.6/site-packages/airflow/contrib/hooks/datadog_hook.py", line 157, in post_event
airflow[9490]: self.validate_response(response)
airflow[9490]: File "/home/airflow/virtualenv/lib/python3.6/site-packages/airflow/contrib/hooks/datadog_hook.py", line 58, in validate_response
airflow[9490]: if response['status'] != 'ok':
airflow[9490]: KeyError: 'status'
但是我的登录都不是在调度程序、网络服务器、工作程序或任务日志中的错误之前或之后。
我已经通过手动导入代码测试了我的 Airflow 工作人员的 datadog_event 调用，并且当我以这种方式运行它时它会正确记录：
airflow@airflow-worker-0:~/analytics$ /home/airflow/virtualenv/bin/python -i /home/airflow/analytics/etl_v2/airflow_data/dags/example_bash_operator_andy.py
[2019-08-07 20:48:01,890] {settings.py:213} INFO - settings.configure_orm(): Using pool settings. pool_size=5, max_overflow=10, pool_recycle=1800, pid=29941
[2019-08-07 20:48:02,227] {__init__.py:51} INFO - Using executor DaskExecutor
>>> datadog_event('My title', 'My task', 'example_bash_operator_andy', 'run_this_last')
[2019-08-07 20:51:17,542] {datadog_hook.py:54} INFO - Setting up api keys for Datadog
[2019-08-07 20:51:17,544] {example_bash_operator_andy.py:38} INFO - My title
[2019-08-07 20:51:17,544] {example_bash_operator_andy.py:39} INFO - My task
[2019-08-07 20:51:17,544] {example_bash_operator_andy.py:40} INFO - example_bash_operator_andy
[2019-08-07 20:51:17,545] {example_bash_operator_andy.py:41} INFO - run_this_last
[2019-08-07 20:51:17,658] {api_client.py:139} INFO - 202 POST https://api.datadoghq.com/api/v1/events (113.2174ms)
我的 airflow.cfg 发布在 https://gist.github.com/andyshinn/d743ddc61956ed7440c500fca962ce92，我使用的是 Airflow 1.10.4。
如何从 DAG 本身输出日志记录或消息以更好地调试可能发生的情况？
【问题讨论】：
-
airflow.cfg 上的 logging_level 是多少？
-
INFO如果有帮助，我可以发布airflow.cfg。除了使用 Dask 作为执行程序和为我的环境设置正确连接的设置之外，它是默认设置。另外，我终于解决了我的主要问题。但是我的问题对于将来从 DAG 进行调试和记录仍然有效。 -
您能否同时添加您正在使用的 Airflow 版本？只是 on_success_callback 还是在其他地方登录都行不通。我能想到的两件事您可能需要检查，1. 您是否在配置 github.com/apache/airflow/blob/master/… 中设置了 logging_config_class。 2.你有remote_logging设置吗，如果有，你那里有数据吗？
-
我使用的是 Airflow 1.10.4。我已编辑问题以添加指向
airflow.cfg的链接。我没有更改任何日志记录。它是默认的（记录到文件，没有远程记录或自定义类）。 -
我认为您可以尝试的一件事是设置 logging_config_class，它现在不是开箱即用的东西。