- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我有一个 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)
知道我如何解决这个问题吗?
最佳答案
第一 CloudSqlInstanceImportOperator
是 deprecated .您应该使用 CloudSQLImportInstanceOperator来自 providersbody
param 需要是字典,如 docs 中所述.
XCOM 是数据库中的一个表。数据保存为字符串。
您不能将 dict 存储在数据库中,因为 dict 是内存对象中的 Python。
您可能有一个 Json(字符串)。尝试将其转换为 dict:
body=json.loads(import_body)
编辑: (在评论中讨论后)
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:
render_template_as_native_obj=True
在您的 DAG 构造函数中。你可以关注这个
documentation例子。
关于python - Airflow xcom pull 只返回字符串,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/64895696/
有没有办法 git pull 直到有东西可以 pull 。 假设我正在等待一位同事推送他的最新更改,因为我不耐烦并且没有看他的屏幕,我只是运行 watch git pull 直到我看到有东西被 pul
这个问题在这里已经有了答案: What is the difference between 'git pull' and 'git fetch'? (37 个答案) 关闭 6 年前。 这个问题我一直
我有一个分支,想使用 $ git pull origin master 将 master 中的更改 git在我这样做之后, pull 确实没有显示任何 merge 的 PR 被 pull 并且说它已经
在各个地方,寻找有关如何从 GitHub merge 请求创建本地分支的说明,我看到了两个版本: git fetch upstream refs/pull/PR_ID/head:NEW_LOCAL_B
我已经在github上创建了一个仓库。 Upstream中有一些新的 pull 请求。我想从本地上游 pull 请求。 我怎样才能做到这一点?我不知道,没有发现任何与此相关的信息。 最佳答案 通过首先
假设我 fork 一个 GitHub 项目并进行一些更改,然后为要 merge 到的更改创建 pull 请求我从中 fork 的原始项目。 一旦我这样做了,如果我返回到我的 fork 项目并提交更多文
我在 BitBucket 上有一个 Git 存储库。我有两个分支 master 和 db。在我的笔记本电脑上,我对 db 进行了更改,提交了更改并推送到 BitBucket。我可以在该分支中看到提交。
我在分支 tm-skeleton并执行 git pull 做某事(从技术上讲,它要求输入密码,然后我在那里点击了 ),但我怎么知道它从哪里 pull ?通常,[branch "tm-skeleton"
我们在组织中使用 Github 进行源代码控制。多个开发人员不断地将他们的更改 merge 到远程源存储库。就我而言,我在两周前克隆了存储库,之后进行了多次 merge 。现在我正在尝试获取代码的最新
在 merge pull 请求之前,我想在本地快速 pull 请求并运行测试并测试一些内容。我还不想点击 gihub merge pull 请求。 我认为滑轮会有所帮助 http://ejohn.or
我有一个服务器,其中有一些我不知道我在哪里的配置我只是 git pull 并获取 github 存储库中的内容,然后重新启动它以进行部署。 问题是,有一个不是我最新的提交,它实际上不在我的服务器上。这
我是 Git 的新手,所以如果这是微不足道的,我深表歉意。我有一个使用 Github 和 EGit 设置的私有(private)存储库. 要更新并 merge 我的本地存储库分支与远程版本(本质上是g
pull 请求的发起者可能 give the maintainer the ability to edit the pull request . 对于任何给定的 pull 请求,我们如何判断该能力是否
我 fork 了一个GitHub存储库,并在 fork 上进行了一些更改,然后提交了一个 pull 请求,但是原始GitHub存储库的所有者要求对它们进行 pull 请求中的一些更改。我以为在fork
如何使用 Git GUI 工具进行拉取?似乎任何地方都没有 pull 命令。 是否有使用 Git GUI 的等效菜单选项? 任何帮助将不胜感激。 最佳答案 嗯,我发现这个有用的论坛帖子: https:
我制作了一个程序来从我的 github 上托管的公共(public)存储库中提取数据。克隆一个项目工作得很好(无论如何我都是这样做的),但我一直在努力让它工作以提取所有内容,一旦你删除一个文件,它不再
哪个比另一个更可取,为什么?如果我在提交之前 pull ,其他开发人员所做的更改是否与我当前正在处理的内容 merge ?如果是这样,那意味着这个更可取? 最佳答案 最好先提交。 pull 而不提交可
我对使用 git 在团队中工作还很陌生。 我从“origin/master”开始。 首先我创建了一个新分支feature/26062018-offline-seite_ef。 我添加了提交。 我检查了
这个问题在这里已经有了答案: Retrieve specific commit from a remote Git repository (11 个答案) 关闭 9 年前。 我在本地创建一个空的新存
我开始在 Github 上使用新的 pull 请求草案功能,但我想知道在我不小心发出 pull 请求的情况下是否可以将合法的 pull 请求转换回 pull 请求草案。 最佳答案 如前所述on Twi
我是一名优秀的程序员,十分优秀!