CODEWITHSUNDEEP
jdbcairflow
Lakshmi2696
Lakshmi2696

Reputation: 13

How to fetch sql query results in airflow using JDBC operator

I have configured JDBC connection in Airflow connections. My Task part of DAG looks like below which contains a select statement. When triggering the DAG is success, but my the query results are not printed in log. How to fetch the results of the query using JDBC operator.

dag = DAG(dag_id='test_azure_sqldw_v1', default_args=default_args,schedule_interval=None,dagrun_timeout=timedelta(seconds=120),)

sql="select count(*) from tablename"

azure_sqldw=JdbcOpetask_id='azure_sqldw',sql=sql,jdbc_conn_id="cdf_sqldw",autocommit=True,dag=dag)

Upvotes: 0

Views: 3083

Answers (1)

Elad Kalif
Elad Kalif

Reputation: 15931

The operator does not print to the log. It just run the query. If you want to fetch results to do something with it you need to use the hook.

from airflow.providers.jdbc.hooks.jdbc import JdbcHook

def func(jdbc_conn_id, sql, **kwargs):
    """Print df from JDBC """
    pprint(kwargs)
    hook = JdbcHook(jdbc_conn_id=jdbc_conn_id)
    df = hook.get_pandas_df(sql=sql,autocommit=True)
    print(df.to_string())


run_this = PythonOperator(
    task_id='task',
    python_callable=func,
    op_kwargs={'jdbc_conn_id': 'cdf_sqldw', 'sql': 'select count(*) from tablename' }, 
    dag=dag,
)

You can also create a custom operator that does the required action you seek.

Upvotes: 1

Related Questions