在Airflow DAG(Directed Acyclic Graphs)中,可以使用以下代码将日志记录到本地时区:
import logging.config
from airflow.utils.log.logging_mixin import LoggingMixin
class MyCustomOperator(LoggingMixin):
def __init__(self, *args, **kwargs):
super(MyCustomOperator, self).__init__()
self.logger.info('Initializing MyCustomOperator...')
def execute(self, context):
self.logger.info('Executing MyCustomOperator...')
# Your code logic here
其中,LoggingMixin是一个实用程序类,可以帮助您在DAG操作中轻松记录日志。该类包含一个名为logger的属性,该属性允许您像往常一样记录日志。如果要将日志记录到本地时区中,则应使用系统默认的logging模块,并在logging配置中设置时区。
# set timezone to local timezone
logging.config.dictConfig({
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'standard': {
'format': '%(asctime)s %(levelname)s %(name)s %(message)s'
},
},
'handlers': {
'airflow': {
'class': 'airflow.utils.log.file_task_handler.FileTaskHandler',
'formatter': 'standard',
'log_filename_template': '{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log',
'base_log_folder': '/usr/local/airflow/logs',
},
},
'loggers': {
'': {
'handlers': ['airflow'],
'level': 'INFO',
'propagate': True,
}
}
})
上一篇:Airflow2.0.1:KubernetesExecutor的Pod模板覆盖未按预期工作
下一篇:Airflow2.0.2-HourlyDAGgettingstuckseeingRefreshingTaskInstancerepeatedly