Airflow -如何从ecs运营商那里推送xcom？
Airflow是一个开源的任务调度和工作流管理平台，用于在云计算环境中管理和调度各种任务和工作流。它提供了一个可视化的用户界面，使用户能够轻松地创建、调度和监控任务和工作流。
在Airflow中，XCom（即交流组件）是一种用于在任务之间传递数据的机制。它允许任务之间共享数据，以便实现任务之间的通信和协作。
要从ECS运营商那里推送XCom，可以按照以下步骤进行操作：
- 首先，确保已经在Airflow中配置了ECS运营商的连接。可以使用Airflow的连接管理界面或配置文件来添加ECS运营商的连接信息，包括访问密钥、密钥ID等。
- 在Airflow中创建一个任务，该任务将从ECS运营商那里获取数据并推送到XCom。可以使用PythonOperator或BashOperator等Airflow提供的Operator来执行任务。
- 在任务中，使用ECS运营商的API或SDK来获取所需的数据。根据ECS运营商的文档和API参考，构建相应的API请求或SDK调用。
- 获取到数据后，使用Airflow提供的XCom API将数据推送到XCom。可以使用
task_instance.xcom_push()方法将数据推送到XCom中，其中task_instance是任务实例的一个引用。 - 在其他任务中，可以使用
task_instance.xcom_pull()方法来获取之前推送到XCom的数据。这样，其他任务就可以使用这些数据进行后续的处理和分析。
推荐的腾讯云相关产品：腾讯云容器服务（Tencent Kubernetes Engine，TKE）。TKE是腾讯云提供的一种容器管理服务，可以帮助用户在云上快速部署、管理和扩展容器化应用。它提供了高可用性、弹性伸缩、安全性等特性，适用于各种规模的应用和工作负载。
产品介绍链接地址：腾讯云容器服务（TKE）
请注意，以上答案仅供参考，具体的实现方式和推荐产品可能因实际需求和环境而有所不同。
相关·内容
在我的airflow dag中，我有一个ecs_operator任务，后跟python操作员任务。我想使用airflow的xcom功能将一些消息从ECS任务推送到python任务。我尝试了do_xcom_push=True选项，但没有结果。在下面的示例dag中找到。 dag = DAG(
dag_name, default_args=default_args, schedule_interval=None)
start = DummyOperator(task_id = 'start'
,dag =dag)
end = Dummy
浏览 10提问于2019-12-17得票数 2
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
回答已采纳
我目前使用的是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
回答已采纳
我尝试通过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
在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
回答已采纳
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
我正在尝试找出将变量从一个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
我对airflow和使用"xcom_push“和"xcom_pull”功能非常陌生。
我有两个dag d1，它有任务t1，第二个dag d2有任务t2。
现在，我使用以下命令从dag d1推送这些值：
kwargs['ti'].xcom_push(key='start_date',value=start_date)
kwargs['ti'].xcom_push(key='end_date',value=end_date)
并使用以下命令在dag d2中拉取相同的start_date和end_date：
start_
浏览 0提问于2019-11-04得票数 3
我刚开始学习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
回答已采纳
在我的airflow spark作业中，我需要将spark作业统计信息传递给工作流中的其他任务。如何将值从SparkSubmitOperator推送到xcom？ task1 = SparkSubmitOperator(
task_id='spark_task',
conn_id='spark_default',
java_class='com.example',
application='example.jar',
name='spark-job',
verbose
浏览 19提问于2019-06-10得票数 2
我正在尝试从CLI触发气流DAG，DAG触发正常，但我收到一堆python警告消息，谁能帮助我，如何抑制这些警告消息。 命令：airflow trigger_dag -e 2021-07-13 Test_CI_DAG 日志： /var/xxx/airflow/venv/lib/python2.7/site-packages/airflow/contrib/operators/ssh_operator.py:65: PendingDeprecationWarning:
Invalid arguments were passed to SSHOperator (task_id: run_spar
浏览 31提问于2021-07-13得票数 0
我正在通过Airflow并行编排多个数据流作业。流水线执行中的任务之一是推送唯一的xcom键来存储每个流水线的临时值，该临时值用于向Bigquery表中插入一个值。数据加载完成后，我需要清除特定的xcom密钥。我不想删除仍在并行运行的其他数据流作业可能生成的其他xcom键。它们是清除特定xcom密钥的方法吗？
浏览 3提问于2021-09-05得票数 0
在运行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
回答已采纳
因此，我用Airflow创建了一个数据流，用于获取存储在中的一些数据，然后将其存储到MongoDB中。因此，我有两个python方法，一个用于获取数据并返回数据，另一个用于将其存储到相关数据库中。如何将一个任务的输出作为另一个任务的输入？到目前为止，这就是我所拥有的(摘要和浓缩版)。
我研究了xcom拉和推的概念，这就是我在下面实现的，我也看到了一个用于气流的MongoHook，但不太确定如何使用它。
import pandas as pd
import pymongo
import airflow
from datetime import datetime, timedelta
from a
浏览 2提问于2019-09-09得票数 0
回答已采纳
我是airflow的新手，有一个关于Airflow及其处理器的问题。当处理器产生输出时，如何将此输出移到下一个处理器的输入中？有一个软件叫做nifi，它将中间输出存储到flowfile中，afaik在airflow中没有这样的东西。那么这是如何发生的呢？
提前谢谢。
浏览 14提问于2017-03-13得票数 10
回答已采纳
我使用的是Airflow 1.8.1，我想推送来自PostgreOperator的sql请求的结果。
以下是我的任务：
check_task = PostgresOperator(
task_id='check_task',
postgres_conn_id='conx',
sql="check_task.sql",
xcom_push=True,
dag=dag)
def py_is_first_execution(**kwargs):
value = kwargs['ti'].
浏览 0提问于2017-08-11得票数 16
回答已采纳
无论我使用xcom_push()推送什么，它总是抛出： WARNING - /usr/local/lib/python3.6/site-packages/sqlalchemy/engine/default.py:588: Warning (1300, "Invalid utf8mb4 character string: '800358'") cursor.execute(statement, parameters) 我知道是xcom_push()抛出了错误，因为如果我注释掉所有的xcom_pushes，那么警告就会消失。我尝试了一个带有静态文本的xcom_pu
浏览 56提问于2020-04-20得票数 0
在Airflow 2 taskflow API中，我可以使用以下代码示例，轻松地在任务之间推送和拉取XCom值： @task(task_id="task_one")
def get_height() -> int:
response = requests.get("https://swapi.dev/api/people/4")
data = json.loads(response.text)
height = int(data["height"])
return h
浏览 17提问于2021-10-03得票数 1
回答已采纳
我是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
我有以下具有两个SSHExecuteOperator任务的DAG。第一个任务执行一个返回参数的存储过程。第二个任务需要此参数作为输入。
请解释一下如何从task1推送的XCom中拉取取值，以便在task2中使用？
from airflow import DAG
from datetime import datetime, timedelta
from airflow.contrib.hooks.ssh_hook import SSHHook
from airflow.contrib.operators.ssh_execute_operator import SSHExecuteOperator
浏览 1提问于2017-03-23得票数 8
回答已采纳
1回答
我是第一次接触气流。我想用气流操作符做一个类似下面的操作。 简而言之，我希望从数据库表中读取一些数据，并根据该表中某列的值执行不同的任务。 这是我用来获取数据的表。 +-----------+--------+
| task_name | status |
+-----------+--------+
| a | 1 |
| b | 2 |
| c | 4 |
| d | 3 |
| e | 4 |
+-----------+--------+ 从上表中，我希望选
浏览 19提问于2020-10-06得票数 1
回答已采纳
我已经搜索了大量的气流博客和文档来调试我的问题。
我想要解决的
检查ftp服务器上是否存在特定文件。
如果它存在的话，把它上传到云
如果不存在，请向客户端发送电子邮件，报告没有找到任何文件。
我有什么
扩展BaseOperator的自定义操作符，它使用SSH并推送一个值(true或false)。
任务，该任务使用BranchPythonOperator从xcom中提取值，并检查前面的任务是否返回true或false，并对下一个任务作出决定。
请看下面的代码。这段代码是我想要做的事情的简化版本。
如果有人对我的原始代码感兴趣，请向下滚动到问题的末尾。
在这里，
浏览 2提问于2019-04-10得票数 3
回答已采纳
有没有办法在DAG步骤的重新运行期间(在清除状态之后)保持XCOM值？ 下面是我试图完成的简化版本，即当DAG步骤状态被清除并重新运行该步骤时，我希望能够加载在上一次运行中推送的XCOM值。但是，即使我可以在XCOM接口中看到该值，也无法提取该值。我查看了pull_xcom()方法的源代码，但不知道它在哪里被过滤掉了。 我试图实现的功能是在DAG的失败运行之间保持一定数量的状态。在本例中，这意味着每次清除并重新运行DAG步骤时，都会将1加到存储值中。 from datetime import datetime
from airflow import DAG
from airflow.ope
浏览 14提问于2020-05-03得票数 1
回答已采纳
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
回答已采纳
在运行DAG时，我会得到上面的错误。
psycopg2.ProgrammingError: can't adapt type 'PythonOperator'
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/taskinstance.py", line
浏览 3提问于2021-09-30得票数 0
回答已采纳
我的用例是，我有一个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
我无法从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
我想使用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
1回答
airflow_version = 1.10.2；python_version = 3.6.8
我很难理解如何使python更可重用到airflow的PythonOperator，因为在dag文件中声明的相同函数可以工作，但是从helper导入它失败了。
因此，下面的工作
def my_function(temp_file, task_id, **kwargs):
xcom_vals = kwargs['ti'].xcom_pull(task_ids=task_id)
if not xcom_vals:
return 'Xcom me
浏览 0提问于2019-03-08得票数 3
回答已采纳
如何解决这个问题？？我不知道该怎么办..。
我的达格斯
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
我试图在subdag创建时从父进程访问一些xcom数据，我在互联网上搜索以实现这一点，但是我没有找到任何东西。
def test(task_id):
logging.info(f' execution of task {task_id}')
def load_subdag(parent_dag_id, child_dag_id, args):
dag_subdag = DAG(
dag_id='{0}.{1}'.format(parent_dag_id, child_dag_id),
default_args=
浏览 0提问于2019-02-18得票数 8
回答已采纳
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
我在代码中定义了以下DAG： from datetime import timedelta, datetime
import airflow
from airflow import DAG
from airflow.operators.docker_operator import DockerOperator
from airflow.contrib.operators.ecs_operator import ECSOperator
default_args = {
'owner': 'airflow',
'retries':
浏览 49提问于2020-08-02得票数 1
回答已采纳
我正在写一个Dag，当我试图从任务中提取Xcom值时，遇到了一个问题。
我想实现这样的目标
{{ dag_run.conf }}中获取手动触发的dag_run参数值的编写函数func
{"dlf_40_start_at":"2022-01-01"，"dlf_40_end_at":"2022-01-02"}，将该值推送到Xcom。
def func(**kwargs):
dag_run_conf = kwargs["dag_run"].conf
task_instance: TaskInstance
浏览 8提问于2022-02-11得票数 1
回答已采纳
我正在做Airflow 1.10。 我在KubernetesPodOperator上运行命令时遇到了问题，在DAG运行时会对整个命令进行评估。 我在DAG运行时生成命令，因为命令的一些参数取决于用户传递的参数。 正如我从文档中读到的，KubernetesPodOperator需要字符串列表或jinja模板列表： :param arguments: arguments of the entrypoint. (templated)
The docker image's CMD is used if this is not provided. 我有PythonOpe
浏览 65提问于2021-08-06得票数 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
回答已采纳
创建了一个包含/airflow/xcom/return.json的镜像，因为日志显示它找不到文件或目录(尝试使用chmod +x) strtpodbefore = KubernetesPodOperator(namespace='rguonew',
image="mydockerimage",
name="fail",
image_
浏览 4提问于2019-03-19得票数 3
我正在使用MySqlHook从airflow_db建立连接，并且正在执行一些查询，但是我需要在某个地方看到查询的结果(假设日志)，如何查看？
这里是示例代码
t1 = MySqlOperator(
task_id='basic_mysql',
mysql_conn_id='airflow_db',
sql="select * from xcom",
dag=dag)
浏览 1提问于2019-01-08得票数 5
回答已采纳
我是个新手，想要创建一个管道，这样赋予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
如何重用在任务之间运行的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
回答已采纳
假设我有一个向XCom推送消息的PythonOperator任务，如何在SparkSubmitOperator中拉取此消息？
def get_some_value(**kwargs):
some_value = 10
return some_value
task1 = PythonOperator(task_id='run_task_1',
python_callable=get_some_value,
provide_context=True,
浏览 2提问于2018-08-21得票数 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
回答已采纳
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
这是我正在尝试的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
我正在尝试实现自定义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
Airflow版本: 1.10.12 我在传递用于下游使用的呈现模板对象时遇到了问题。我正在尝试从Airflow conf中获取两个配置变量。 from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime
with DAG(
dag_id="example_trigger_target_dag",
default_args={"owner": "airflow"},
浏览 8提问于2021-06-29得票数 1
我的气流代码有下面的Python操作符可调用，我在这里创建一个列表并将其推送到xcoms：
keys = []
values = []
def attribute_count_check(e_run_id,**context):
job_run_id = int(e_run_id)
da = "select count (distinct row_num) from dds_metadata.dds_temp_att_table where run_id ={}".format(job_run_id)
cursor.execute(da)
浏览 8提问于2022-10-01得票数 0
我有一份气流的工作如下：
import time
job_id = int(time.time())
airflow_job1 = PythonOperator(op_kwargs={"job_id" : job_id}, ...)
airflow_job2 = BashOperator(op_kwargs={"job_id" : job_id}, ...)
airflow_job1 >> airflow_job2
我知道每次脚本启动时，我都会有一个新的job_id，用于每个气流任务。但我想知道，如果我从中间运行脚本，比如airflow_job
浏览 2提问于2020-05-28得票数 0
回答已采纳
问题摘要：
我需要使用xcom从一个SSHOperator中获得标准输出。
筛选一些行并获取输出值，以便将它们传递给另一个SSHOperator
不幸的是，我没有在气流文件中找到任何有用的东西
代码示例：
import airflow
from airflow.operators.dummy_operator import DummyOperator
from airflow.contrib.operators.ssh_operator import SSHOperator
default_args = {
'owner': 'airflow&
浏览 9提问于2020-07-31得票数 1
1回答
气流动态任务返回列表而不是字典
、、、、
请参考下面的代码。
当recon_rule_setup任务运行时，每次它都从以前的read_recon_config任务获得字典(recon_conf)作为输入。
但是，当recon_rule_exec运行时，它将从以前的任务中获得列表作为输入(recon_rule)。
我的期望是，根据返回值，recon_rule_setup应该运行2次，recon_rule_exec应该运行4次。
为什么每次扩张都有不同的表现。
from datetime import datetime
from airflow.models import DAG, XCom
from airflow.utils.date
浏览 5提问于2022-11-03得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
