如何在日志中查看 MySqlHook 结果 [英] How to see MySqlHook result in log

问题描述

我正在使用 MySqlHook 从 airflow_db 建立连接，我正在执行一些查询，但我需要在某处查看查询结果(比如日志)，我怎么看?

这是示例代码

I am using MySqlHook to establish connection from airflow_db, and I am performing some query, but I need to see the result of the query somewhere ( let say log), how can I see?

Here is the sample code

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

推荐答案

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

The MySQL operator currently (airflow 1.10.1 at time of writing) doesn't support returning anything in XCom, so the fix for you, for now, is to write a small operator yourself. You can do this directly in your DAG file:

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

这篇关于如何在日志中查看 MySqlHook 结果的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！