我打算使用PostgreSQL作为我的任务元信息提供者,因此我想运行一些查询并获取一些数据,然后将其作为填充变量传递到另一个任务。
问题是当我使用PostgresHook时,我可以获得数据,但它在一个Python方法中,我无法访问,事实上我看到下面这行代码:
[2021-08-23 13:00:12,628] {python.py:151} INFO - Done. Returned value was: [[1, "inf_account",....]]
这是我的部分代码:
def _query_postgres(**context):
"""
Queries Postgres and returns a cursor to the results.
"""
postgres = PostgresHook(postgres_conn_id="aramis_postgres_connection")
conn = postgres.get_conn()
cursor = conn.cursor()
mark_williams = cursor.execute(" SELECT * FROM public.aramis_meta_task; ")
# iterate over to get a list of dicts
details_dicts = [doc for doc in cursor]
# serialize to json string
details_json_string = json.dumps(details_dicts, default=json_util.default)
task_instance = context['task_instance']
task_instance.xcom_push(key="my_value", value=details_json_string)
return details_json_string
但是我不知道应该使用哪个变量来访问它,也不知道如何将其推送到XCOM中,以便我可以将返回的值作为参数传递给另一个BashOperator任务(例如Spark)。
另一方面,PostgresOperator仅返回None
作为结果。