在自定义操作员气流中拉取xcom值
是指在Apache Airflow中使用XCom机制来传递数据。XCom是Airflow中的一种机制，用于在任务之间传递数据。它允许任务之间共享数据，以便实现更复杂的工作流逻辑。
XCom值可以在任务之间传递，包括从一个任务的输出传递到另一个任务的输入。这对于需要在任务之间共享数据的情况非常有用。通过使用XCom值，可以实现任务之间的数据依赖性和通信。
在自定义操作员气流中拉取XCom值的步骤如下：
- 在一个任务中，使用
ti.xcom_push()方法将数据推送到XCom。 - 在另一个任务中，使用
ti.xcom_pull()方法来拉取先前推送的数据。
以下是一些使用XCom值的优势和应用场景：
优势：
- 灵活性：XCom值允许任务之间动态地共享数据，使工作流更加灵活和可扩展。
- 数据传递：通过使用XCom值，可以将数据从一个任务传递到另一个任务，实现任务之间的数据依赖性。
- 任务通信：XCom值还可以用于任务之间的通信，使任务能够共享状态和信息。
应用场景：
- 数据传递：当一个任务的输出数据需要作为另一个任务的输入数据时，可以使用XCom值来传递数据。
- 任务通信：当任务之间需要共享状态或信息时，可以使用XCom值进行任务间通信。
- 动态参数：当任务的参数需要根据其他任务的结果进行动态调整时，可以使用XCom值来传递参数。
腾讯云相关产品和产品介绍链接地址：
- 腾讯云Apache Airflow：腾讯云提供了托管的Apache Airflow服务，可用于构建和管理工作流。详情请参考：腾讯云Apache Airflow
- 腾讯云云函数（Serverless）：腾讯云云函数是一种无服务器计算服务，可用于编写和运行无需管理服务器的代码。它可以与Apache Airflow结合使用，实现更灵活的工作流。详情请参考：腾讯云云函数
- 腾讯云消息队列（CMQ）：腾讯云消息队列是一种高可靠、高可用的消息队列服务，可用于在任务之间传递数据。它可以与Apache Airflow结合使用，实现任务之间的数据传递。详情请参考：腾讯云消息队列
相关·内容
我已经在我的气流DAG中创建了一个任务，它返回一些值，这个值应该可以在xcom部分中看到，但是它不是。我设置了do_xcom_push = True，但它仍然没有显示出来。在此之前，我使用的是气流2.0.0，与将返回值推送给xcom的任务相同，但在气流2.1.2中，它没有发生。我不明白我错过了什么。下面是气流任务的一个片段：
任务= python_operator.PythonOperator(task_id="invoke_cf"，python_callable=invoke_cloud_function，do_xcom_push=True)
浏览 1提问于2021-10-19得票数 0
回答已采纳
我理解，通过在xcom_push=True中设置SimpleHttpOperator，我可以从访问从Xcom返回的数据。但我不太清楚我是如何做到的。这是通过创建带有回调的PythonOperator并在回调中调用xcom_pull实现的吗？一些示例代码将是很好的看到。
我想要做的是将Google的股价数据作为CSV数据读取，然后将它们插入到MySQL数据库中，以了解气流。一开始，我想我只需要使用两个操作符：
SimpleHttpOperator >> MySqlOperator
但是现在我想我必须在中间添加PythonOperator？：
SimpleHttpOperator &g
浏览 0提问于2018-02-25得票数 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
回答已采纳
我们正在开发ETL工具，使用apache和apache气流。Apache气流将用于工作流管理。
能处理大量的数据吗？可以从apache气流中提取、转换计数吗？
浏览 1提问于2021-08-02得票数 0
我对气流中的xcoms很陌生，无法获得准确的信息，我们有一个ssh操作符任务，如下所示，使用"echo“命令。
task_id = SSHOperator(
task_id="task_id",
ssh_conn_id="CONNECTION_ID",
command="echo Failed"
)
是否有任何方法将上面的任务输出捕获到以下变量
variablea = "Failed"
如有任何建议请。我试过xcom push=True，但它显示的是空白值。
浏览 12提问于2022-09-20得票数 1
回答已采纳
我有3次DAG测试：
DAGR 1 2019-02-13 16:00执行
DAGR 2于2019年-02-13:00执行
DAGR 3于2019-02-13 :00执行
在任务实例X of DAGR 1中，我希望获得任务实例Y的xcom值。我做了这个：
kwargs['task_instance'].xcom_pull(task_ids='Y')
我期望从DAGR 1中的任务实例DAGR 1中获得xcom的值。相反，我是从DAGR 3得到的。
从气流文件
如果xcom_pull被传递给task_ids的单个字符串，则返回来自该任务的最新XC
浏览 1提问于2019-02-13得票数 1
我已经构建了一个气流操作符，它针对一个云API执行HTTP请求。在执行一个请求之后，我得到一个run_id来标识执行，然后继续检查我的请求状态，直到它完成为止(可能需要一些时间)。
我正在尝试开发一种机制，将此run_id保存到xCom，以便在我的吊舱与云服务的连接松散时，解决网络问题/ spot问题。我正在使用气流重试，所以我希望在失败后恢复相同的连接，并继续检查保存到xCom的xCom的状态。
我能够将run_id推到xCom，但是当我为相同的任务提取它时，我得到的是None，而不是我在Admin仪表板中推送和看到的值(我可以将它拉到正在执行的其他任务中)。
是否对从我试图从同一任务中推送
浏览 1提问于2021-04-25得票数 1
回答已采纳
目前，我有一个带有Python运算符的气流中的DAG，相关的python可以这样调用：
def push_xcom(**kwargs):
ti = kwargs["ti"]
ti.xcom_push(key=key, value=value)
xcom_opr = PythonOperator(
task_id='xcom_opr',
python_callable=push_xcom,
dag=dag
)
这个dag的目标是更新在气流中定义的其他DAG的xcom变量。这不可能吗？我找不
浏览 2提问于2019-12-04得票数 2
回答已采纳
我有一个http端点列表，每个端点都执行自己的任务。我们正在尝试编写一个应用程序，该应用程序将通过以特定顺序调用这些端点来进行编排。在这个解决方案中，我们还必须处理一个http端点的输出，并为下一个http enpoint生成输入。此外，根据触发器，可以同时调用相同的工作流。 到目前为止，我所做的是: 1.定义了一个从HttpOperator派生的新操作符，并引入了将http端点的输出写入文件的功能。2.我写了一个python运算符，它可以根据需要的逻辑来传输输出。 因为我可以执行同一工作流的多个实例，所以我不能对输出文件名进行硬编码。有没有办法使我写的http操作符写入一些唯一的文件名，并
浏览 28提问于2019-12-31得票数 2
1回答
使用BashOperator从气流DAG任务中触发ansible剧本，因为我想获取在task 1中生成的值，并使用那些在气流DAG任务2中生成的值。
任务1执行生成此值的ansible剧本。在气流DAG任务2中是否有获得这个值的方法?遵循下面的doc，但是找不到涵盖ansible剧本用例的任何示例代码。
浏览 3提问于2022-09-23得票数 0
1回答
我在xcom中的一个气流任务中保存了xcom中的值(key='db_value')。在后面的一个步骤中，我需要在Bigquery表中保存相同的值。请您建议，如何准备SQL更新/插入表，并在string中添加XCOM值，然后将其传递给BashOperator、PythonOperator或BigqueryOperator。我尝试使用python或BigqueryOperator，但是由于task_instance.xcom_pull在执行运算符之前是不可用的，所以我无法预先准备好task_instance.xcom_pull，以便在param中提供相同的内容。
任何建议。
浏览 1提问于2022-06-27得票数 1
因此，我用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
回答已采纳
1回答
我试图使用气流运行11步的AWS电子EMR，并遵循这个作为参考。由于在11个步骤中使用EmrAddStepsOperator和EmrStepSensor，所以会重复太多。所以我试着翻阅它。我在DAG中使用了下面的代码。
step_adder = list()
step_checker = list()
steps = ['step1', 'step2', 'step3', 'step4', 'step5', 'step6'...till step11]
# @evalcontextfilter
#
浏览 5提问于2021-02-03得票数 2
回答已采纳
2回答
清理气流数据库。Xcom表
、、、、
在气流中清洁xcom表的最佳方法是什么？它运行在postgres db的码头上。
我尝试用查询(从xcom删除)删除一些数据，并尝试运行以下引用：，但是它不工作，xcom表的大小仍然没有减少
与本例相关，它占用了我的主机服务器上的大量存储空间。
编辑:第一版使用:气流1.10.3
浏览 32提问于2022-04-25得票数 0
回答已采纳
我想对这个sql文件运行sqlfluff命令。此SQL是针对雪花的，将在气流中执行。ti.xcom_pull是一种用于气流数据采集的函数。
DELETE FROM test
WHERE date = '{{ti.xcom_pull(key='filedate')}}';
然后我在fix命令中得到了这个错误。
L: 1 | P: 1 | TMP | Unrecoverable failure in Jinja templating: 'ti' is
| undefined. Have
浏览 11提问于2022-09-12得票数 1
回答已采纳
我有一个自定义操作符，它按如下方式推送XCOM值：
...
task_instance = context['task_instance']
task_instance.xcom_push("list_of_files",file_list)
...
效果很好。我有一个definition (my_dag.py)，在这里我使用自己的操作符创建一个任务，它推送XCOM值，然后使用这个xcom值来执行in循环。怎么拉？
浏览 0提问于2017-07-20得票数 6
回答已采纳
我试图从SSHOperator获得param到Xcom，并在python中获得它。
def decision_function(**context):
ti = context['ti']
output_ssh= ti.xcom_pull(task_ids='ssh_task')
print('ssh output is: {}'.format(output_ssh))
ls = SSHOperator(
task_id="ssh_task",
command= "ls&#
浏览 9提问于2022-06-15得票数 0
1回答
我对气流很陌生，我觉得我可能错过了一些惯例或概念。
上下文:我定期将文件放到S3桶中。我的管道需要抓取新文件并处理它们。
基本上:如何避免再处理？
将来管道的某些部分也不太可能发生变化，I将希望重新处理文件。但在日常工作中我不想再处理文件。此外，将来可能还会有其他管道，它们需要从一开始就开始，并为不同的输出处理所有的文件。
我有很多保存状态(本地json文件，或检查输出文件的存在)的方法，但是我想知道在气流中是否有约定。对我来说，目前最有意义的是重用为气流存在的postgres (可能是糟糕的表单？)，添加另一个DB并开始在其中创建表，其中我列出了输入文件，如果它们已经为工作流X、工作流Y等进
浏览 2提问于2017-12-06得票数 4
1回答
我正在使用气流自定义操作符和来自任务id的xcom值(动态生成)。
如何使用xcom拉取动态任务id的值。
for id in ids :
def dummy_push_function(**context):
context['ti'].xcom_push(key='some_id', value='abc')
dummy_operator = DummyOperator(
task_id='Start',
dag=main_dag
)
push_fun
浏览 0提问于2022-04-11得票数 1
回答已采纳
我用Apache气流写了一个DockerOperator，我想给它一个卷。到目前一切尚好。下面是一个示例：
t = DockerOperator(
task_id='test',
image='testimage:latest',
command='python3 /code/test.py',
volumes=["/mnt/interim:/interim"],
xcom_push=True,
dag=dag,
)
我的问题是：
挂载
浏览 1提问于2020-01-02得票数 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
我们有一些跨越多个AWS区域的应用程序。因此，我们不需要部署多个气流来处理我们的ETL任务(每个区域一个)，我们想知道是否有办法在不同的区域/集群/名称空间中部署工作人员。
我们的气流部署运行在EKS中，所以我猜这可能是KubernetesPodOperator中的一个设置(如果有的话)。我也没有找到通过DAG指定集群的方法，但我希望这里的一些天才有一个想法。
谢谢你，比尔
浏览 6提问于2021-09-27得票数 1
我的用例是，我有一个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回答
我试图在气流2+中创建一个DAG，它将使用CloudDqtaFusionStartPipeline操作符触发多个数据融合管道，它们将并行运行。
但是，我希望分配参数值(如管道名称、运行时参数等)。对于每个数据融合管道，基于以前Python任务的输出动态。
我正在尝试的流程如下所示。
开始- read_bq - df_1，.df_n
这里，read_bq是一个BigQuery任务，它将从BigQuery表中读取值作为列表(管道名称、运行时参数等值)。
然后循环遍历这个列表，我将确定要触发多少个数据融合管道，并将这些管道返回的值赋值给这些管道。
我面临的问题是，CloudDqtaFusionSta
浏览 8提问于2022-10-16得票数 0
我是airflow的新手，有一个关于Airflow及其处理器的问题。当处理器产生输出时，如何将此输出移到下一个处理器的输入中？有一个软件叫做nifi，它将中间输出存储到flowfile中，afaik在airflow中没有这样的东西。那么这是如何发生的呢？
提前谢谢。
浏览 14提问于2017-03-13得票数 10
回答已采纳
假设我有一些气流操作符，给操作符的一个参数需要从xcom获取值。我设法做到了以下几点-
f"model_id={{{{ ti.xcom_pull(task_ids='train', key='{task_id}')}}}}"
其中，model_id是码头运算符的参数名，气流运行，task_id是xcom中该值的键名。
现在，我想做一些更复杂的事情，在task_id下保存一个字典，而不是一个值，并且能够从它中拿出它。
有没有类似于我上面提到的方法呢？就像-
f"model_id={{{{ ti.xcom_pull(task_ids=&#
浏览 2提问于2022-06-28得票数 0
2回答
假设DAG包含两个任务：DAG: Task A >> Task B (BashOperators或DockerOperators)。他们需要通过XComs进行沟通。
Task A通过stdout中的一行json输出信息，然后可以在Task A的日志中检索这些信息，因此可以在其return_value XCom键if xcom_push=True中检索。例如：{"key1":1,"key2":3}
Task B只需要来自Task A的key2信息，所以我们需要反序列化return_value XCom of Task A来提取这个值并使用jinja模板
浏览 17提问于2022-10-15得票数 0
我的DAG中有一些部分会生成列表，我不能将这些列表分解为单独的任务，以便在下游单独处理。 下面是一个伪例： def push(**kwargs):
# """Pushes an XCom without a specific target"""
for n in range(10):
kwargs['ti'].xcom_push(key=f'vals', value=n)
def puller(**kwargs):
ti = kwargs['ti']
浏览 16提问于2019-02-26得票数 1
1回答
我对气流有点陌生，在运行一段时间后，我似乎无法找到一种删除所有Xcom的方法。我找到了bash命令airflow清除，但是这个已删除的任务实例也被删除了，我只想清除Xcom。欢迎任何帮助。谢谢!
更新：
clear_xcom_operator = SqliteOperator(
task_id="clear_Xcom",
sqlite_conn_id="sqlite_default",
sql="delete from xcom where dag_id='dumper'",
retries=0,
浏览 2提问于2020-06-24得票数 0
在气流1.10.10DAG中，我们有一个ShortCircuitOperator，它使用python函数check_remote_server()来决定分支。
在check_remote_server_data()函数中，我们如何启动到远程服务器的SSH连接，在其上运行bash命令并获得结果？
是否可以使用我以前使用Web定义的气流SSH连接？
def check_remote_server_data():
pass # how can we use a predefined Airflow SSH connection, named `remote`?
with dag:
浏览 1提问于2020-07-31得票数 0
回答已采纳
我们在k8s中有很多长时间运行的内存/cpu密集型作业，它们都是在google平台上的kubernetes上与芹菜一起运行的。然而，我们在交付的scaling/retrying/monitoring/alerting/guarantee方面有很大的问题。我们想从芹菜转向更先进的框架。
有一个比较：，但这还不够。
气流：
具有较好的支持社区~400 vs ~12标签SO，13k星vs ~3.5k星
python定义流的方式比使用yamls更好
作为产品的GCP支持:云编写器
更好的仪表板
一些不错的操作员，比如电子邮件操作员
Argoproj：
对Kubernet
浏览 0提问于2019-07-15得票数 8
我试着在气流中运行电子病历，并找到上面写的例子。
step_adder = EmrAddStepsOperator(
task_id='add_steps',
job_flow_id="{{ task_instance.xcom_pull(task_ids='create_job_flow', key='return_value') }}",
aws_conn_id='aws_default',
steps=SPARK_STEPS,
)
浏览 14提问于2022-04-03得票数 1
回答已采纳
是否可以在气流模板中解析JSON字符串？
我有一个HttpSensor，它通过REST监视作业，但是作业id在上游任务的响应中，该任务被xcom_push标记为True。
但是，下面的代码会给出错误的jinja2.exceptions.UndefinedError: 'json' is undefined
t1 = SimpleHttpOperator(
http_conn_id="s1",
task_id="job",
endpoint="some_url",
method='POST&#
浏览 2提问于2017-11-27得票数 8
回答已采纳
我看到了很多关于如何在气流中使用xcom_push和xcom_pull与PythonOperators的例子。
我需要从一个非PythonOperator类中执行xcom_pull，但是无法找到如何实现它。
任何指针或示例都将不胜感激！
浏览 3提问于2018-07-29得票数 10
回答已采纳
我有一个气流运算符，它返回一个字符串值，任务名为'task1‘。因此，在执行之后，我进入xcom并检查return_value和它只是一个字符串(下面是截图)。
然后，我有一个名为task2的操作符，它从task1获取xcom值的输入，如下所示：
"{{ ti.xcom_pull(task_ids=['task1'],key='return_value')}}"
问题是它得到的值是一个转换为字符串的列表。
值在xcom中：--这只是一个字符串
‘这只是一个字符串’
那么，是否有一种方法，我可以更新xcom拉(静亚版本)，上面显示的只是提
浏览 3提问于2021-08-17得票数 1
回答已采纳
我有一个简单的气流DAG：
from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.bash import BashOperator
with DAG("Second Dag",
start_date=datetime(2022,1,1),
schedule_interval="0 5 * * *",
catchup=False,
max_active_runs=1
浏览 14提问于2022-02-10得票数 1
1回答
我有一个SSHOperator，它可以为stdout编写文件路径。我希望获得该文件路径的os.path.basename，以便将它作为参数传递给我的下一个任务(即sftp拉)。其想法是将远程文件下载到当前的工作目录中。到目前为止，这就是我所拥有的：
with DAG('my_dag',
default_args = dict(...
xcom_push = True,
)
) as dag:
# there is a
浏览 22提问于2022-06-21得票数 0
在气流图形用户界面中，可以看到我的xcom变量看起来像['val1', 'val2', 'val3']。如果我从文档中正确地理解了它，那么当我完成时，{{ ti.xcom_pull("task_id") }}将是一个字符串。因此，我的问题是如何拆分xcom varible，以便获得三个独立的值，然后在我的其他任务中使用？
浏览 5提问于2022-06-22得票数 0
1回答
我是第一次接触气流。我想用气流操作符做一个类似下面的操作。 简而言之，我希望从数据库表中读取一些数据，并根据该表中某列的值执行不同的任务。 这是我用来获取数据的表。 +-----------+--------+
| task_name | status |
+-----------+--------+
| a | 1 |
| b | 2 |
| c | 4 |
| d | 3 |
| e | 4 |
+-----------+--------+ 从上表中，我希望选
浏览 19提问于2020-10-06得票数 1
回答已采纳
我们有一个kubernetes pod操作符，它将吐出python字典，该字典将定义哪个下游的kubernetes pod操作符与它们的依赖项一起运行，以及传递给每个操作符的环境变量。
如何将这个python字典对象返回到executor的上下文(或者它是worker的上下文？)这样气流就能孕育出下游的kubernetes操作者？
我看过BranchOperator和TriggerDagRunOperator，XCOM推/拉，Variable.get和Variable.set，但是似乎什么都没有用。
浏览 1提问于2019-03-12得票数 2
1回答
我的DAG (我们称之为DAG_A)使用trigger_dagrun运算符启动另一个DAG (DAG_B)。DAG_B的任务使用XCOM，我希望在完成后从DAG_B的运行任务中获得XCOM值(这正是我已经启动的任务)。
使用XCOM并不是一个困难的要求--基本上，气流本身提供的任何(合理的)机制都能工作。如果需要，我可以更改DAG_B。
找不到任何这样的例子，所以感谢你的帮助。
计划B是让DAG_B将XCOM值保存到一些持久存储中，比如DB或文件以及一些运行的id，DAG_A将从那里获取它。但是，如果有一些内置的机制，我想避免这种复杂的情况.
浏览 3提问于2021-04-28得票数 7
1回答
我们正试图将每日(CSV)提取的文件合并到数据仓库中。
在我们的用例中，DAG的python代码对于我们所有的DAG都是相同的(~2000)，因此我们通过DAG生成器逻辑从一个文件生成它们。在我们的DAG中，我们只有15个任务(5个虚拟任务，2个CloudDataFusionStartPipelineOperator任务，8个python任务)。
在DAG生成过程中，我们读取气流变量(~30-50)来确定要生成哪些DAG(这也决定了DAG的ID以及它们应该处理的模式/表名称)。我们称之为发电机变量。
在DAG生成过程中，DAG还通过ID读取它们的配置(每个生成的DAG多出2-3个气流变量)。我
浏览 4提问于2022-04-10得票数 2
回答已采纳
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
回答已采纳
我试图在气流中设置AWS过程，我需要EmrCreateJobFlowOperator中的EmrCreateJobFlowOperator和EmrAddStepsOperator中的steps通过位于其他地方的单独JSON文件来设置。
我尝试过许多方法，既可以直接链接JSON文件，也可以设置和获取JSON的气流变量。如果我要使用气流变量，它们也需要动态命名，这是我遇到的问题。我可以使用Variable.set轻松地使用PythonOperator实现动态名称，但由于气流在PythonOperator之外编写Python代码的局限性，无法在job_flow_overrides或steps中使用动态
浏览 1提问于2019-09-26得票数 2
我试图使用xcom_pull在气流中动态地提取值。
当我将op_kwargs中的f-字符串拼凑在一起时，下面提到的格式并不适用于我。如果有人能帮我的话我很感激。
op_kwargs={'names':"{{ ti.xcom_pull(key = '" + f'name{i+1}' + ", task_ids='places' ) }}"}
浏览 12提问于2022-11-18得票数 1
我是新的气流自动化，我现在不可能做到这一点与apache气流(或luigi等)，或我应该做一个长的bash文件来这样做。
我想要为这个构建守护程序
在AWS EMR上创建/克隆集群
安装python要求
安装与火花放电相关的储气罐
从github获取最新代码
提交火花作业
在finish上终止群集
对于各个步骤，我可以像下面这样制作.sh文件(不确定这样做是否好)，但不知道如何在气流中这样做。
1)使用cluster.sh创建cluser
aws emr create-cluster \
--name "1-node dummy cluster
浏览 1提问于2019-04-16得票数 2
回答已采纳
我从数据库中获取了一个值，并将其存储在XCom中，我想用1来增加它。是否有可能增加存储在XCom中的值？
'{{ ti.xcom_pull("task_id") + 1}}'
'{{ int(ti.xcom_pull("task_id")) + 1}}'
编辑
这是我的气流DAG的一部分。我有一项任务是从Hbase中提取数据：
pull_data_hbase = BashOperator(
task_id='pull_data_hbase',
dag=dag,
bash_command=&
浏览 3提问于2022-01-13得票数 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
回答已采纳
2回答
(我知道，但它已经足够老了，我现在问它，希望有一个最新的答案。)
我正在为气流为AWS批处理编写插件，在作业通过AWSBatchOperator操作符提交到批处理之后，我需要一种方法来等待作业完成并通知气流。我正在为它编写一个传感器，因为它似乎是处理它的适当方法。
但是，当AWSBatchOperator在AWS批处理中创建作业时，它需要向传感器提交作业id，以便传感器知道要检查什么作业。是否有任何简单的方法将信息从已完成的任务传递到其依赖的任务？我知道XCom，但考虑到这种情况的简单性，以及我想象的传感器常见问题，这似乎有点过火了。
谢谢!
浏览 20提问于2017-01-20得票数 2
我试图设置将数据从一个任务传递到另一个任务。
然而，在葡萄园套接字运行了一段时间之后，我面对的是vineyard._C.NotEnoughMemoryException，这是讨论过的
尽管有默认的args作为
default_args = dict(
owner = DEFAULT_OWNER,
depends_on_past = False,
start_date = days_ago(1),
on_failure_callback = partial(notify_failure_and_cleanup_xcom, users=["fip"]
浏览 12提问于2022-03-05得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
