没有这样的文件或目录/airflow/xcom/ directory .json
这个错误信息是指在路径"/airflow/xcom/"下找不到".json"文件或目录。根据这个错误信息，可以得出以下答案：
- 错误信息解释： 这个错误信息表明在指定的路径下找不到名为".json"的文件或目录。
- 解决方法：
- 确认路径是否正确：首先，需要确认路径"/airflow/xcom/"是否正确。检查路径是否存在拼写错误或者是否存在其他错误。
- 检查文件或目录是否存在：确认在指定路径下是否存在名为".json"的文件或目录。如果不存在，可以尝试创建该文件或目录。
- 检查权限：确保当前用户对该路径及其下的文件或目录具有足够的权限。
- 相关概念：
- 文件：在计算机中，文件是存储数据的一种方式。它可以包含文本、图像、音频、视频等各种类型的数据。
- 目录：目录是文件系统中用于组织和存储文件的一种结构。它可以包含文件和其他目录，形成一个层次结构。
- JSON：JSON（JavaScript Object Notation）是一种轻量级的数据交换格式，常用于前后端数据传输和存储。它使用易于阅读和编写的文本格式，表示结构化的数据。
- 应用场景：
- 文件和目录：在计算机系统中，文件和目录广泛应用于数据存储、文件管理、软件开发等各个领域。
- JSON：JSON常用于Web开发中的数据传输和存储，例如前后端数据交互、API接口返回数据等。
- 腾讯云相关产品：
- 对象存储（COS）：腾讯云对象存储（COS）是一种安全、高可用、低成本的云端存储服务，适用于存储和处理大规模非结构化数据。 产品链接：https://cloud.tencent.com/product/cos
请注意，由于要求不提及特定的云计算品牌商，因此无法提供其他品牌商的产品链接。
相关·内容
我正在尝试实现自定义XCOM后端。
这些是我所做的步骤：
在主气流dir (AIRFLOW_HOME).Created中创建“”目录--这些"custom_xcom_backend.py“文件位于：
from typing import Any
from airflow.models.xcom import BaseXCom
import pandas as pd
class CustomXComBackend(BaseXCom):
@staticmethod
def serialize_value(value: Any):
if isinsta
浏览 9提问于2022-02-11得票数 2
在kubernetes集群中放置了一个Apache Airflow实例: webserver、scheduler和postgresql。使用基于bitnami的自定义舵图，并做了一些更改。 Airflow正在与KubernetesExecutor一起工作。我所有的DAG都是PythonOperator和KubernetesPodOperator (以前的DockerOperator -在k8s之前)。Xcom推送只有在PythonOperator下才能正常工作，但在KubernetesPodOperator中，我在执行结束时得到了错误(所有dags都会受到影响)： [2019-12-06 1
浏览 91提问于2019-12-06得票数 3
回答已采纳
创建了一个包含/airflow/xcom/return.json的镜像，因为日志显示它找不到文件或目录(尝试使用chmod +x) strtpodbefore = KubernetesPodOperator(namespace='rguonew',
image="mydockerimage",
name="fail",
image_
浏览 4提问于2019-03-19得票数 3
在运行DAG时，DAG使用码头映像运行jar，
给出了xcom_push=True，它将在单个吊舱中创建另一个容器以及码头映像。
达格：
jar_task = KubernetesPodOperator(
namespace='test',
image="path to image",
image_pull_secrets="secret",
image_pull_policy="Always",
node_selectors={"d-type":"na-node-g
浏览 1提问于2018-11-27得票数 5
回答已采纳
我的用例是，我有一个S3事件，它触发一个lambda (在一个S3 createobject事件上)，它反过来调用一个气流DAG，传入两个-conf值(桶名，文件键)。
然后，我将使用Python操作符提取键值，并将其存储在xcom变量中。然后，我希望在S3ToSnowflakeOperator中提取这个xcom值，并实质上将该文件加载到雪花表中。
流程的所有部分都是工作栏，在S3ToSnowflakeOperator任务中提取xcom值。我的日志中基本上有以下内容。
查询:从@MYSTAGE_.SOURCE_PARAMS_JSON_DEMO/files=复制到“原始”ti.xcom({ti.
浏览 1提问于2022-03-29得票数 1
1回答
我发现xcom实际上将数据写入数据库并从其他任务中提取数据。我的数据集很大，对它进行分类和写入数据库会造成一些不必要的延迟。是否有一种不使用xcom在同一气流Dag中的任务之间通信数据的方法？
下面是我尝试过的代码，上下文实际上没有被传递。我知道使用task_instance.xcom_push()可以工作，但它也会对数据进行筛选，并将其写入我不需要的数据库中。
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, t
浏览 4提问于2020-02-18得票数 3
如何重用在任务之间运行的DAG上计算的值？
我试图在我的DAG中生成一个时间戳，并将它用于几个任务中。到目前为止，我尝试设置一个变量和一个params值-没有什么工作，它是唯一的每一个任务运行。
这是我的代码：
from datetime import datetime, timedelta
from airflow import DAG
from airflow.models import Variable
from airflow.utils.dates import days_ago
from airflow.providers.amazon.aws.operators.athena
浏览 0提问于2021-03-08得票数 0
回答已采纳
如何解决这个问题？？我不知道该怎么办..。
我的达格斯
def _game():
game_df = pd.DataFrame(columns=['match', 'puuid']
input_data = {'match':match100,'puuid':game_content['puuid']}
game_df = game_df.append(input_data, ignore_index=True)
return game_df
def _game_score():
浏览 18提问于2022-09-19得票数 0
我是Apache气流的新手，我试图编写我的第一个Dag，它有一个基于另一个任务的任务(使用ti.xcom_pull)
PS :我使用VScode在WSL Ubuntu20.04中运行气流。
我创建了一个任务1 (task_id = "get_datetime")，该任务运行"date“bash命令(它可以工作)
然后我创建了另一个任务(task_id='process_datetime')，它占用第一个任务的日期时间并对其进行处理，我设置了python_callable，一切都很好。
问题是，当我在终端中运行“气流任务测试(ti.xcom_pull ta
浏览 12提问于2022-11-04得票数 0
我是airflow的新手，有一个关于Airflow及其处理器的问题。当处理器产生输出时，如何将此输出移到下一个处理器的输入中？有一个软件叫做nifi，它将中间输出存储到flowfile中，afaik在airflow中没有这样的东西。那么这是如何发生的呢？
提前谢谢。
浏览 14提问于2017-03-13得票数 10
回答已采纳
1回答
在气流任务之间传递熊猫数据帧failing.Tried下面的代码
try:
from datetime import timedelta, datetime
from airflow import DAG
from airflow.operators.python import PythonOperator
import pandas as pd
except Exception as e:
print(e)
default_args = {
"owner": "airflow",
"start
浏览 4提问于2021-05-15得票数 0
大家好，所以我的达格实际运行良好，所有的输出都在工作，但气流的UI不改变为成功，并由于以下问题失败。我在网上读到了以下两篇文章：
do_xcom_push=False
在气流版本2.0中，Xcom_push将被废弃。
我只是不知道该怎么安排这个？有谁能分享你的见解吗？
完整代码：
import pandas as pd
import logging
import csv
import numpy as np
import datetime
import glob
import shutil
import time
import gcsfs
from airflow.operators impo
浏览 14提问于2021-12-15得票数 3
1回答
气流版本:2.0.2试图通过重试AWS机密管理器的数据来创建Emr集群。
我正在尝试编写一个气流守护程序，我的任务是从这个get_secret函数中获取数据，并在Spark_steps中使用它。
def get_secret():
secret_name = Variable.get("secret_name")
region_name = Variable.get(region_name)
# Create a Secrets Manager client
session = boto3.session.Session()
clien
浏览 5提问于2022-06-11得票数 1
回答已采纳
2回答
我有以下两个文件。一个带有DAG和两个任务(DummyOperator和TaskGroup)。
# example_dag.py
from datetime import timedelta
from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from includes.taskgroup import build_
浏览 8提问于2021-12-14得票数 0
目标
目标是获得一个非常简单的气流码头，我只是想安装几个非常基本的pip软件包从requirements.txt。Tan运行了一个包含numpy和熊猫部件的python脚本。
命令
docker build -t my38 .
docker-compose up airflow-init
docker-compose up -d
文件
气流/码头文件
FROM apache/airflow:latest-python3.8
COPY requirements.txt .
RUN pip install -r requirements.txt
气流/要求
apache-airflow==2.4.
浏览 8提问于2022-09-23得票数 0
回答已采纳
这是我正在尝试的KubernetesPodOperator的一个例子-- set_tag = KubernetesPodOperator(
namespace='default',
task_id='set-tag',
name='set-tag',
image='ubuntu:18.04',
xcom_push=True,
cmds=["/bin/sh", "-c"],
arg
浏览 29提问于2020-07-01得票数 0
我目前正在使用Airflow和BigQuery操作符来触发各种SQL脚本。当SQL直接写入Airflow DAG文件时，这种方法可以很好地工作。例如：
bigquery_transform = BigQueryOperator(
task_id='bq-transform',
bql='SELECT * FROM `example.table`',
destination_dataset_table='example.destination'
)
但是，我想将SQL存储在一个单独的文件中，保
浏览 36提问于2019-07-24得票数 2
回答已采纳
我想使用airflow composer将最新的文件从gcs存储桶复制到本地。我试图使用gustil cp来获取最新的文件并加载到本地airflow中，但遇到了问题: CommandException:没有URL匹配错误。如果我检查对象，我会得到XCom =‘Objects’.Any建议？
download_file = BashOperator(
task_id='download_file',
bash_command="gsutil cp $(gsutil ls -l gs://<bucket_name> | sort -k 2 | tai
浏览 0提问于2020-11-13得票数 1
我刚开始学习Airflow，但是很难掌握Xcom的概念。因此我写了这样一个dag：
from airflow import DAG
from airflow.utils.edgemodifier import Label
from datetime import datetime
from datetime import timedelta
from airflow.operators.bash import BashOperator
from airflow.contrib.operators.ssh_operator import SSHOperator
from airflow.co
浏览 14提问于2021-06-11得票数 2
回答已采纳
1回答
我是第一次接触气流。我想用气流操作符做一个类似下面的操作。 简而言之，我希望从数据库表中读取一些数据，并根据该表中某列的值执行不同的任务。 这是我用来获取数据的表。 +-----------+--------+
| task_name | status |
+-----------+--------+
| a | 1 |
| b | 2 |
| c | 4 |
| d | 3 |
| e | 4 |
+-----------+--------+ 从上表中，我希望选
浏览 19提问于2020-10-06得票数 1
回答已采纳
在Airflow中，我有一个JSON customer_schema，我想将其转换为一个名为final_schema.json的XCom文件并上传到Google Cloud Storage中。我在谷歌云存储中的存储桶名为northern_industrial_customer。我尝试使用下面的FileToGoogleCloudStorageOperator，但它不起作用。 有人知道如何将我的XCom任务ID customer_schema作为名为final_schema.json的JSON文件传输到谷歌云存储吗 transfer_to_gcs = FileToGoogleCloudStora
浏览 30提问于2019-09-15得票数 0
1回答
我安装了气流，使用对接-合成和运行db init命令。我正在尝试创建一个使用DockerOperator执行某些脚本的DAG。在我的DockerOperator中，脚本试图读取气流变量并使用BaseHook获取连接。但是，我的DockerOperator中的脚本似乎连接到(空的) sqlalchemy数据库，而不是通过UI设置填充连接和变量的初始化postgres数据库。
是否有一种方法可以将DockerOperator上下文提供给包含通过UI设置的连接和变量的气流数据库？
我的气流DAG
from datetime import datetime
from json import dumps
浏览 6提问于2022-06-02得票数 1
回答已采纳
1回答
在我的气流DAG中，我的任务之一是将数据从BQ导出到文件(我在URI中使用通配符* )，并且我看到在导出到文件的特定任务结束时，它会启动DAG的额外运行，在第一个任务中失败。
运行的次数似乎取决于它导出的文件数量(当我将URI更改为没有通配符时，它就在成功导出之后启动了一个额外的运行)。
为什么会发生这种情况？为了避免这些多次运行，我可以做什么不同的事情呢？
这就是它在气流树中的样子：
import datetime
import json
from airflow import utils
from airflow.models import DAG
from airflow.op
浏览 9提问于2022-10-20得票数 0
回答已采纳
我尝试使用我的自定义http操作符(扩展simpleHttpOperator )在我的http请求的headers值中使用模板。似乎只在数据字段中支持模板。如何在headers字段中实现同样的功能。我想要传递模板化的授权头。请在下面找到我的代码。 import airflow
from airflow import DAG
from airflow.configuration import conf
from airflow.operators.python_operator import PythonOperator
from airflow.operators.auth_plugins
浏览 22提问于2019-06-25得票数 0
回答已采纳
下面是我的Dag:我正在尝试旋转一个EMR集群，并尝试在添加步骤时使用它的集群id。用例是:想要旋转一个集群并将其集群id保存在s3中的某个地方。但是xcom_pull显示出了错误。
from datetime import timedelta
from airflow import DAG
from airflow.contrib.operators.emr_add_steps_operator import EmrAddStepsOperator
from airflow.providers.amazon.aws.operators.emr_create_job_flow import E
浏览 3提问于2021-12-02得票数 1
我正在尝试使用Airflow的上传一些二进制数据。
在使用python的时，我只需使用
requests.put(f'my.url.example',
data=open('path_to_file, 'rb'), # parameter containing path to file
auth=(username, password))
但是，使用我的Airflow的操作符，在调用open()时，文件路径不能正确呈现
put_task = SimpleHttpOperator(
t
浏览 1提问于2021-03-10得票数 0
1回答
我使用的气流1.10.2叉从流行的和使用CLI操作它读取。我已经创建了一个DAG，当我通过UI触发它时，它将被成功地执行。但是，当我使用命令行触发DAG时，DAG状态将开始运行，但任务将永远不会执行。
对于下面图像的引用，DAG的ID是xcom_4_modules_ver_5
在通过调度的进程运行DAG并通过UI手动触发之后，我在CLI上执行airflow list_dag_runs xcom_4_modules_ver_5以检查DAG运行的状态。DAG已成功执行，并将预期的输出写入日志文件中。这对应于ID 1和2。
但是，在执行命令airflow trigger_dag -sd /
浏览 1提问于2019-04-15得票数 8
我看到了一些关于它的类似问题(比如和)，但它们都没有回答这个问题。
我想用BashOperator运行一些python文件。如下所示：
my_task = BashOperator(
task_id='my_task',
bash_command='python3 /opt/airflow/dags/programs/my_task.py',
)
有没有办法从my_task.py调用xcom_push和xcom_pull
浏览 1提问于2021-08-03得票数 0
我尝试通过airflow cli test命令测试两个任务`
第一个任务运行时，auto将最后一个控制台推送到xcom，我在airflow GUI中看到了预期的值some value
当我通过airflow cli test命令运行第二个任务时，我只是得到了None作为返回值，但正如我在这里读到的：它应该工作，至少xcom_push是明显工作的，为什么不是xcom_pull？
有人给出了如何让它工作的提示吗？Provide context设置为true。
示例代码：
t1 = BashOperator(
task_id='t1',
bash_command=&#
浏览 16提问于2019-11-15得票数 0
def f_get_value_via_xcom(**context):
var = context['ti'].xcom_pull(dag_id='bash_execute', key='return_value')
bash_execute = BashOperator(
task_id='bash_execute',
bash_command='some bash command'
xcom_push=True,
priority_weight=100,
浏览 2提问于2021-01-22得票数 2
对于xcom_pull，我面临一个奇怪的问题，它总是返回一个xcom_pull字符串"{{ task_instance.xcom_pull(dag_id = 'cf_test',task_ids='get_config_val',key='http_con_id') }}"
我的要求很简单，我使用python运算符推动了xcom，使用xcom_pull，我试图检索该值并将其作为http_conn_id传递给SimpleHttpOperator，但是变量返回一个字符串，而不是解析xcom_pull值。Python Operator能
浏览 2提问于2022-05-29得票数 0
回答已采纳
我目前正在遵循这里给出的模板：来创建一个DAG来调用一个emr实例，使用submit。在设置spark_test_steps时，我需要包括从POST Json中传入的变量，以填充火花提交，如下所示：
SPARK_TEST_STEPS = [
{
'Name': 'calculate_pi',
'ActionOnFailure': 'CONTINUE',
'HadoopJarStep': {
'Jar': 'com
浏览 3提问于2019-09-19得票数 0
回答已采纳
1回答
运行deps时会出现此错误。我是不是在dockerfile中遗漏了一些东西来提供对dbt_modules的访问？我似乎找不到它的位置，甚至在文档中也找不到dbt_modules。我为几个yml文件提供了代码。提前感谢
回溯
2022-06-07 04:34:59.121970 (MainThread): Running with dbt=0.21.1
2022-06-07 04:34:59.972911 (MainThread): You have an incompatible version of 'pyarrow' installed (6.0.1), please in
浏览 7提问于2022-06-07得票数 1
回答已采纳
我无法从Python连接到Airflow的元数据数据库。
连接已设置，我可以使用UI的即席查询窗口查询元数据数据库。
如果我尝试使用相同的连接，但从Python，它将不会工作。
所以，详细地说，我有两个连接设置，都是从UI: connection 1: Conn type: MYSQL Host: airflow-sqlproxy-service Schema: composer-1-6-1-airflow-1-10-0-57315b5a登录: root
连接2:连接类型: MYSQL主机: 127.0.0.1架构: composer-1-6-1-airflow-1-10-0-57315b5a
浏览 0提问于2020-11-27得票数 1
我是个新手，想要创建一个管道，这样赋予Python的函数就会接受变量，并使用XCOM方法，将这些变量交给Bash。使用Bash，我尝试运行整个python脚本，它需要从XCOM方法获取的这些变量。有人能帮我创建这样的管道吗？
我正在附加下面的坏代码。在下面的代码中，app.py是采用count_check和recon_check值的python脚本。另外，一个bash命令可以提取多个变量吗？
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bas
浏览 5提问于2022-06-22得票数 0
我正在尝试执行一个由几个Python函数组成的气流脚本。这些函数基本上是查询数据库并执行很少的任务。我正在尝试执行这是气流，这样我就能够单独监控每一个功能。给出的代码是我试图执行的代码，并得到以下错误
子任务: NameError:名称“task_instance”未定义为
## Third party Library Imports
import psycopg2
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime
浏览 3提问于2018-05-30得票数 1
回答已采纳
我正在为动态任务生成创建python类。对于第一种情况，我接受来自用户的输入，并将该输入传递给动态任务创建部分，但无法在函数内部更新类变量。在下面的代码1中，我调用accept_ip_cli，并期望这个函数将更新self.my_dict变量，但是当我试图在test_dict_value()中读取self.my_dict时，我得到的是空字典。
class my_class:
def __init__(self,
airflow_cfg, dag,
default_args,
*arg
浏览 4提问于2021-12-31得票数 0
回答已采纳
1回答
我对气流非常陌生，并且阅读了它的大部分文档。从文档中，我了解到DAG中的组件之间的小数据可以使用XCom类共享。DAG中发布数据的组件必须推送，而订阅数据的组件必须拉出。
然而，我不太清楚的语法部分推和拉。我指的是关于XCom的部分，并开发了一个代码模板。假设我有下面的代码，它只有两个组件，一个推手和一个拉杆。推送器发布拉出器必须消耗并写入日志文件的当前时间。
from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
lo
浏览 1提问于2019-04-05得票数 0
回答已采纳
关于this earlier question，假设我们有一个Apache Airflow DAG，它包含两个任务，首先是一个HTTP请求(即SimpleHTTPOperator)，然后是一个处理第一个任务的响应的PythonOperator。 为了方便起见，以Dog CEO API为例，考虑以下DAG： from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.http.operators.http import SimpleHttpOperator
from airfl
浏览 48提问于2021-10-30得票数 2
回答已采纳
1回答
我最近发现了气流，我想做几个简单的例子来了解它是如何工作的。所以我构建了一个非常简单的DAG (example_airflow.py)，如下所示
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import timedelta
def first_task(**kwargs):
a = 1+1
print(a)
kwargs['ti'].xcom_push(key='
浏览 8提问于2022-07-05得票数 0
我正在尝试找出将变量从一个BashOperator任务传递到另一个任务的最通用/最有效的方法。我想出了一个解决方案，将第一个BashOperator任务的输出推送到xcom。然后，该字符串由PythonOperator任务提取，该任务将该字符串解析为键值对，然后将其推送到xcom。最后，这些k-v对可以被第二个BashOperator任务拉取和使用。 我想从更有经验的airflow用户那里得到一些评论，这种方法是不是太复杂了，还是还可以？ from airflow import DAG
from airflow.operators.bash_operator import BashOpera
浏览 74提问于2019-10-14得票数 0
TL;DR如何使dag中的任务可以访问资产(SQL文件、文本文件 大家好， 我正在运行单节点Airflow安装(1.10.12)。我有一个使用SQL语句的dag。出于可读性的考虑，我希望将这些语句保存在它们自己的SQL文件中。所以我的项目看起来是这样的： MyProject
- airflow_ftp_DAG.py
- airflow_ftp_assets (directory)
- WEX_HH_Upsert_SQL.sql 当我的dag运行时，它会抱怨找不到SQL文件。这是因为dag在不同的文件夹中运行它的任务(如运行os.getcwd()所示) 如何使我的S
浏览 7提问于2020-10-20得票数 1
回答已采纳
如何使用SimpleHttpsOperator读取之前消息的XCom，然后决定在Airflow中执行任务2。 假设我有3个SimpleHttpsOperator任务，所有任务都返回一条XCom消息，在XCom值中，它根据结果返回成功或失败。 所以在执行t2之前，我想检查一下t1是否成功。我的所有任务都使用SimpleHttpsOperator t1 >> t2 >> t3 下面是我的代码片段： t1 = SimpleHttpOperator(
task_id='t1',
htt
浏览 3提问于2019-04-04得票数 0
我需要向对象DatabricksRunNowOperator().传递一个job_id参数。job_id是执行databricks jobs create --json '{myjson}命令的结果。
$ databricks jobs create --json '{myjson}'
{job_id: 12}
import os
import subprocess
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.c
浏览 3提问于2019-09-16得票数 2
回答已采纳
我目前使用的是Airflow (版本: 1.10.10)， 我对创建DAG很感兴趣，它将每小时运行一次， 这将带来Docker容器的使用信息(磁盘使用)， (可通过docker CLI命令( df -h)获得的信息)。 我的理解是：“如果xcom_push为真，那么当bash命令完成时，写入标准输出的最后一行也将被推送到XCom。” 但我的目标是从bash命令中获得一个特定值，而不是最后编写的一行。 例如，我想得到这一行(参见screeeshot) "tmpfs 6.2G 0 6.2G 0% /sys/fs/cgroup“ 转换为我的XCOM值，这样我就可以编辑和提取其中的特定值，
浏览 40提问于2021-10-31得票数 0
回答已采纳
当我用Dockerfile为airflow exporter构建二进制文件和打包成高山图像时，我得到了错误。
不确定如何修复此错误。
docker生成+++++++++时出现+++++错误
---> Running in caebfe9a04a0
stat mage.go: no such file or directory
The command '/bin/sh -c cd /go/src/github.com/airflow_exporter/; go run mage.go binary' returned a non-zero code: 1
+++++++++
浏览 2提问于2018-10-09得票数 0
我正在我的airflow项目中使用PythonVirtualenvOperator，我需要将参数从一个任务传递到另一个任务，以测试我使用这个的xcoms，它可以工作，但当我将pythonOperator更改为PythonVirtualenvOperator时，它有一个问题。
代码：
from airflow import DAG
from airflow.operators.python_operator import PythonOperator, PythonVirtualenvOperator
args = {
'owner': 'Airflow'
浏览 48提问于2019-12-02得票数 2
2回答
我试图从XCOM变量生成一组动态任务。在XCOM中，我存储一个列表，我希望使用列表中的每个元素动态地创建一个下游任务。
我的用例是，我有一个上游操作符，它检查sftp服务器上的文件，并返回符合特定条件的文件名列表。我希望为返回的每个文件名创建动态下游任务。
我把它简化到下面，当它工作的时候，我觉得它不是一个惯用的气流解决方案。在我的用例中，我会编写一个python函数，这个函数是从python操作符中调用的，它从xcom中提取值并返回它，而不是使用pusher函数。
我明白，虽然我可以创建一个自定义操作符，将两者结合在一起，但我并不认为创建一个丢弃操作符是一个好做法，我希望还有另一个解决方案。
浏览 0提问于2019-04-14得票数 14
回答已采纳
我希望将日期值(来自变量)作为sql文件中的分区列传递。变量没有得到评估。变量存储在Xcom中。有没有更好的方法。
渲染模板：
TO 's3://opsake/prc/deldb/ldp_pipeline_hist/partiton_key=''{{ ti.xcom_pull(task_ids='genExecParam', key='app_run_dt') }}''/'
iam_role 'arn:aws20xxx:role/Re11dccc3role'
allowoverw
浏览 3提问于2022-07-05得票数 0
回答已采纳
DockerOperator有一个参数xcom_push，设置该参数时，会将Docker容器的输出推送到Xcom：
t1 = DockerOperator(task_id='run-hello-world-container',
image='hello-world',
xcom_push=True, xcom_all=True,
dag=dag)
在Xcom下的管理界面中，我可以使用键return_value看到这些值。但是，如何在DAG中访
浏览 3提问于2019-11-20得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
