Issue
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)
Solution
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
Answered By - kaxil
0 comments:
Post a Comment
Note: Only a member of this blog may post a comment.