如何解决如何使用 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,)
您还可以创建一个自定义运算符来执行您寻求的所需操作。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。