gpt4 book ai didi

python - Airflow xcom pull 只返回字符串

转载 作者:行者123 更新时间:2023-12-04 00:53:03 24 4
gpt4 key购买 nike

我有一个 Airflow 管道,我需要从 pubsub 订阅中获取文件名,然后将该文件导入到云 sql 实例中。我使用 CloudSqlInstanceImportOperator 导入 CSV 文件。该运算符需要一个主体,其中包含文件名和其他参数。由于我在运行时读取了该文件名,因此我还必须在运行时定义主体。这一切都有效。但是当我从 xcom 拉出正文时,它返回一个字符串而不是 python 字典。所以 CloudSqlInstanceImportOperator 给了我以下错误(我的猜测是,因为正文是字符串而不是字典):

Traceback (most recent call last)
File "/usr/local/lib/airflow/airflow/models/taskinstance.py", line 984, in _run_raw_tas
result = task_copy.execute(context=context
File "/usr/local/lib/airflow/airflow/contrib/operators/gcp_sql_operator.py", line 715, in execut
self._validate_body_fields(
File "/usr/local/lib/airflow/airflow/contrib/operators/gcp_sql_operator.py", line 712, in _validate_body_field
api_version=self.api_version).validate(self.body
File "/usr/local/lib/airflow/airflow/contrib/utils/gcp_field_validator.py", line 420, in validat
dictionary_to_validate=body_to_validate
File "/usr/local/lib/airflow/airflow/contrib/utils/gcp_field_validator.py", line 341, in _validate_fiel
value = dictionary_to_validate.get(field_name
AttributeError: 'str' object has no attribute 'get
这是我使用的代码:
import json 
import os
from datetime import datetime, timedelta
import ast
from airflow import DAG
from airflow.contrib.operators.gcs_to_gcs import GoogleCloudStorageToGoogleCloudStorageOperator
from airflow.operators.python_operator import PythonOperator
from airflow.operators.bash_operator import BashOperator
from airflow.contrib.sensors.pubsub_sensor import PubSubPullSensor
from airflow.contrib.sensors.gcs_sensor import GoogleCloudStoragePrefixSensor
from airflow.operators.dagrun_operator import TriggerDagRunOperator
from airflow.contrib.operators.gcp_sql_operator import CloudSqlInstanceImportOperator


def create_dag(dag_id,default_args):
BUCKET = "{{ var.value.gp2pg_bucket }}"
GCP_PROJECT_ID = "{{ var.value.gp2pg_project_id }}"
INSTANCE_NAME = "{{ var.value.gp2pg_instance_name }}"

def define_import_body(file,**kwargs):
import_body = {
"importContext": {
"importUser": "databasename",
"database": "databaseuser",
"fileType": "csv",
"uri": "bucketname" + file,
"csvImportOptions": {
"table": "schema.tablename",
"columns": ["columns1",
"column2"]}
}
}
task_instance = kwargs['task_instance']
task_instance.xcom_push(key='import_body', value=import_body)
print(import_body)

def get_filename(var,**kwargs):
message = ast.literal_eval(var)
file = message[0].get('message').get('attributes').get('objectId')
task_instance = kwargs['task_instance']
task_instance.xcom_push(key='filename', value=file)
print(file)

dag = DAG(dag_id=dag_id, schedule_interval=None, default_args=default_args)

with dag:
t1 = PubSubPullSensor(task_id='pull-messages',
project="projectname",
ack_messages=True,
max_messages=1,
subscription="subscribtionname")


message = "{{ task_instance.xcom_pull() }}"

t2 = PythonOperator(
task_id='get_filename',
python_callable=get_filename,
op_kwargs={'var': message},
provide_context=True,
)

file = "{{ task_instance.xcom_pull(task_ids='get_filename', key='filename') }}"

t3 = PythonOperator(
task_id='define_import_body',
python_callable=define_import_body,
op_kwargs={'file': file},
provide_context=True,
)

import_body = "{{ task_instance.xcom_pull(task_ids='define_import_body', key='import_body') }}"

t4 = CloudSqlInstanceImportOperator(
project_id=GCP_PROJECT_ID,
body= import_body,
instance=INSTANCE_NAME,
gcp_conn_id='postgres_default',
task_id='sql_import_task',
validate_body=True,
)

t5 = GoogleCloudStorageToGoogleCloudStorageOperator(
task_id='copy_files',
source_bucket=BUCKET,
source_object=file,
destination_bucket=BUCKET,
destination_object='processed/import/'+file, )

t1 >> t2 >> t3 >> t4 >> t5

return dag


dags_folder = os.getenv('DAGS_FOLDER', "./dags")
flow_config = open(f'{dags_folder}/gp2pg/flow_config.json', 'r').read()
for key, values in json.loads(flow_config).items():
default_args = {
"owner": "owner",
"start_date": datetime(2020, 1, 1),
"email": [],
"email_on_failure": False,
"email_on_retry": False,
"retries": 0,
"retry_delay": timedelta(minutes=5),
}

dag_id = f"gp2pg_{key}_data_to_pg"

globals()[dag_id] = create_dag(dag_id, default_args)
知道我如何解决这个问题吗?

最佳答案

第一 CloudSqlInstanceImportOperatordeprecated .您应该使用 CloudSQLImportInstanceOperator来自 providersbody param 需要是字典,如 docs 中所述.
XCOM 是数据库中的一个表。数据保存为字符串。
您不能将 dict 存储在数据库中,因为 dict 是内存对象中的 Python。
您可能有一个 Json(字符串)。尝试将其转换为 dict:

body=json.loads(import_body) 
编辑: (在评论中讨论后)
您需要使用 PythonOperator 包装您的运算符,以便您可以转换 xcom听写并使用它。
def my_func(ds, **kwargs):
ti = kwargs['ti']
body = ti.xcom_pull(task_ids='privious_task_id')
import_body = json.loads(body)
op = CloudSqlInstanceImportOperator(
project_id=GCP_PROJECT_ID,
body=import_body,
instance=INSTANCE_NAME,
gcp_conn_id='postgres_default',
task_id='sql_import_task',
validate_body=True,
)
op.execute()


p = PythonOperator(task_id='python_task', python_callable=my_func)
编辑:对于 Airflow >= 2.1.0:
Airflow 添加了将字段呈现为原生 Python 对象的功能。
您需要设置 render_template_as_native_obj=True在您的 DAG 构造函数中。你可以关注这个 documentation例子。

关于python - Airflow xcom pull 只返回字符串,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/64895696/

24 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com