Airflow中PythonOperator的日志记录和任务监控

在Airflow中，PythonOperator是用于在DAG中执行Python函数的Operator。PythonOperator提供了日志记录和任务监控的功能，可以通过两种方式实现。

一种方式是使用Python的logging模块进行日志记录。在Python函数中，可以使用logging模块的方法打印日志。这些日志将会显示在任务日志中，并且可以在Airflow的Web UI中查看。下面是一个示例代码：

import logging

def my_task():
    logging.info("Task started")
    
    # task logic
    
    logging.info("Task completed")

task = PythonOperator(
    task_id='my_task',
    python_callable=my_task,
    dag=dag,
)

在上面的代码中，我们使用logging.info()方法记录了任务的开始和结束。这些日志将会显示在任务的运行日志中。

另一种方式是使用Airflow自带的log模块进行日志记录。这种方式与上面的方式类似，但是使用了Airflow的log模块的方法，其日志级别和配置更加灵活。下面是一个示例代码：

from airflow import AirflowLogHandler, logging

def my_task():
    logger = logging.getLogger('my_task')
    logger.addHandler(AirflowLogHandler())
    
    logger.info("Task started")
    
    # task logic
    
    logger.info("Task completed")

task = PythonOperator(
    task_id='my_task',
    python_callable=my_task,
    dag=dag,
)

在上面的代码中，我们首先获取了一个名为'my_task'的logger，并添加了AirflowLogHandler。然后，我们使用logger的方法记录了任务的开始和结束。

除了日志记录，PythonOperator还提供了任务监控的功能。我们可以在Python函数的适当位置使用Airflow的BaseOperator的方法来记录任务的状态。下面是一个示例代码：

from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults

class MyPythonOperator(BaseOperator):
    
    @apply_defaults
    def __init__(
        self,
        *args, **kwargs):
        super(MyPythonOperator, self).__init__(*args, **kwargs)
    
    def execute(self, context):
        self.log.info("Task started")
        
        try:
            # task logic
            
            self.log.info("Task completed successfully")
            return "success"
        except Exception as e:
            self.log.error("Task failed: %s", str(e))
            raise
    
    def on_kill(self):
        self.log.warning("Task killed")

task = MyPythonOperator(
    task_id='my_task',
    dag=dag,
)

在上面的代码中，我们定义了一个继承自BaseOperator的自定义Operator，并重载了execute()方法。在execute()方法中，我们使用self.log.info()方法记录了任务的开始和完成信息。在on_kill()方法中，我们使用self.log.warning()方法记录了任务被杀死的信息。

通过以上方式，我们可以在Airflow中实现任务的日志记录和任务监控。这些日志和状态将会在任务的运行日志和任务的元数据中展示和记录，方便任务的开发和监控。