如何利用JDBC操作符在airflow中提取sql查询结果

Question

我已经在Airflow Connections中配置了JDBC连接。DAG的My Task部分如下所示，其中包含一个select语句。当触发DAG是成功的，但我的查询结果不打印在日志中。如何使用JDBC操作符获取查询的结果。

dag=dag(DAG_ID='test_azure_sqldw_v1',default_args=default_args,schedule_interval=none,dagrun_timeout=timedelta(Seconds=120))

sql=“从表名中选择计数(*)”

azure_sqldw=jdbcopetask_id='azure_sqldw',sql=sql,jdbc_conn_id=“cdf_sqldw”,autocommit=true,dag=dag)

Answer 1

操作员不打印到日志。它只是运行查询。如果您想要获取结果来执行某些操作，您需要使用钩子。

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,
)

还可以创建执行所需操作的自定义运算符。