如何使用 JDBC 运算符在气流中获取 sql 查询结果

问题描述

我已经在 Airflow 连接中配置了 JDBC 连接。 DAG 的我的任务部分如下所示，其中包含一个选择语句。当触发 DAG 成功时，但我的查询结果未打印在日志中。如何使用 JDBC 运算符获取查询结果。

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)

解决方法

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

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

您还可以创建一个自定义运算符来执行您寻求的所需操作。