时间戳

首页 > 解决方案 > 如何在日志中查看 MySqlHook 结果

python-3.x - 如何在日志中查看 MySqlHook 结果

问题描述

我正在使用MySqlHook建立从airflow_db的连接，并且我正在执行一些查询，但是我需要在某处查看查询结果（比如说日志），我怎样才能看到？

这是示例代码

t1 = MySqlOperator(
    task_id='basic_mysql',
    mysql_conn_id='airflow_db',
    sql="select * from xcom",
    dag=dag)

标签： python-3.xairflow

解决方案

MySQL 运算符当前（撰写本文时为 airflow 1.10.1）不支持在 XCom 中返回任何内容，因此目前对您的修复是自己编写一个小型运算符。您可以直接在 DAG 文件中执行此操作：

from airflow.operators.python_operator import PythonOperator
from airflow.operators.mysql_operator import MySqlOperator
from airflow.hooks.mysql_hook import MySqlHook

class ReturningMySqlOperator(MySqlOperator):
    def execute(self, context):
        self.log.info('Executing: %s', self.sql)
        hook = MySqlHook(mysql_conn_id=self.mysql_conn_id,
                         schema=self.database)
        return hook.get_records(
            self.sql,
            parameters=self.parameters)

t1 = ReturningMySqlOperator(
    task_id='basic_mysql',
    mysql_conn_id='airflow_db',
    sql="select * from xcom",
    dag=dag)

def get_records(**kwargs):
    ti = kwargs['ti']
    xcom = ti.xcom_pull(task_ids='basic_mysql')
    string_to_print = 'Value in xcom is: {}'.format(xcom)
    # Get data in your logs
    logging.info(string_to_print)

t2 = PythonOperator(
    task_id='records',
    provide_context=True,
    python_callable=get_records,
    dag=dag)

t1 >> t2

推荐阅读