我刚开始学习Airflow，但是很难掌握Xcom的概念。datetimefrom airflow.contrib.operators.ssh_operatorimport SSHOperator
from airflow.contrib.hooks.ssh_hook import SSHHoo
浏览 14提问于2021-06-11得票数 2
回答已采纳
我试图在循环中创建多个任务，并在BashOperator和SSHOperator中传递BashOperator和SSHOperator中动态生成的任务ids。= PostgresOperator( postgres_conn_id='pg_airflow_dbload_hist_pkg_to_ts_'+str(i),
浏览 0提问于2019-07-22得票数 2
回答已采纳
问题摘要：
代码示例：from airflow.operators.dummy_operatorimport DummyOperator
from airflow.contrib.operators.ssh_operator imp
浏览 9提问于2020-07-31得票数 1
我试图从SSHOperator获得param到Xcom，并在python中获得它。def decision_function(**context): output_ssh= ti.xcom_pull(task_ids='ssh_task')
ls = SSHOperator</e
浏览 9提问于2022-06-15得票数 0
我正在尝试从CLI触发气流DAG，DAG触发正常，但我收到一堆python警告消息，谁能帮助我，如何抑制这些警告消息。to SSHOperator (task_id: run_spark_job).Invalid arguments were: *args: () **kwargs: {'xcom_push': True}*args: ()
**kwargs: {'xcom
浏览 31提问于2021-07-13得票数 0
我正在检索第一个任务实例中的密码，我需要用第二个任务实例中的密码更新SSH连接，并在第三个任务实例中使用它。t1 :调用R函数来检索svc帐户的密码(存储在xcom_push中)
目前t1和t2按预期工作，但是t3失败了，因为密码没有更新，它正在寻找基于.ssh密钥文件的身份验证下面是我的</
浏览 0提问于2018-07-06得票数 3
SSHOperator返回打印的最后一行，在本例中是"remote_IP“。但是，SSHOperator的返回值是使用UTF-8编码的。Read_remote_IP = SSHOperator( ssh_hook=hook,)
浏览 3提问于2021-06-13得票数 0
回答已采纳
我正在尝试找出将变量从一个BashOperator任务传递到另一个任务的最通用/最有效的方法。我想出了一个解决方案，将第一个BashOperator任务的输出推送到xcom。然后，该字符串由PythonOperator任务提取，该任务将该字符串解析为键值对，然后将其推送到xcom。最后，这些k-v对可以被第二个BashOperator任务拉取和使用。我想从更有经验的airflow用户那里得
浏览 74提问于2019-10-14得票数 0
我想使用airflow composer将最新的文件从gcs存储桶复制到本地。我试图使用gustil cp来获取最新的文件并加载到本地airflow中，但遇到了问题: CommandException:没有URL匹配错误。如果我检查对象，我会得到XCom =‘Objects’.Any建议？download_file = BashOperator( bash_command
浏览 0提问于2020-11-13得票数 1
例如：若要更新可能在dag定义中使用的日期值，请执行以下操作。 task_id='task_1_%s' % tabletable[&qu
浏览 1提问于2019-10-24得票数 0
回答已采纳
我目前使用的是Airflow (版本: 1.10.10)， 我对创建DAG很感兴趣，它将每小时运行一次， 这将带来Docker容器的使用信息(磁盘使用)， (可通过docker CLI命令( df -我的理解是：“如果xcom_push为真，那么当bash命令完成时，写入标准输出的最后一行也将被推送到XCom。” 但我的目标是从bash命令中获得一个特定值，而不是最后编写的一行。例如，我想得到这一行(参见screee
浏览 40提问于2021-10-31得票数 0
回答已采纳
job_id是执行databricks jobs create --json '{myjson}命令的结果。import DAGfrom airflow.contrib.operators.databricks_operatorDatabricksRunNowOperator
returned_output = subprocess.check_o
浏览 3提问于2019-09-16得票数 2
回答已采纳
1回答
我在计划中为远程ssh上的run命令创建了这个dag。from datetime import timedeltafrom airflow.operators.bash_operator import BashOperatorfrom airflow.contrib.operators.ssh_operator import SSHOperator
'owner': &#x
浏览 3提问于2021-05-22得票数 0
我看到了一些关于它的类似问题(比如和)，但它们都没有回答这个问题。my_task = BashOperator( bash_command='python3 /opt/airflow/dags/programs/my_task.py',有没有办法从my_task.p
浏览 1提问于2021-08-03得票数 0
我的locale文件中有一个python脚本，我不想将其SCP到远程机器上，并通过airflow远程触发的SSHOperator运行。如何在远程计算机上运行区域设置.py文件并获得结果？我需要SSHOperator和python_callable，而不是bash_command。 有人能给我看一下像SSHPYTHONOperator这样的远程自定义操作符示例吗？我解决问题的方法如下： gettime="""
import
浏览 18提问于2020-12-22得票数 1
我试着用气流来处理成千上万的文件。为了在我的任务之间共享这些文件/数据，我目前只是保存和从sql服务器中提取数据。然而，我相信使用XComs会容易得多，并允许我更多地抽象我的代码。但是，我知道在使用Xcoms的任务之间可以共享多少数据是有限制的。对于这个限制是否只涉及通过一个Xcom发送大数据，或者是否也适用于通过许多Xcoms发送大量数据(每个xcom只包含4kb的数据)，我感到困惑。
浏览 6提问于2022-07-19得票数 0
回答已采纳
Airflow版本: 1.10.12 我在传递用于下游使用的呈现模板对象时遇到了问题。我正在尝试从Airflow conf中获取两个配置变量。from airflow import DAG schedule_interval='@once',
浏览 8提问于2021-06-29得票数 1
我试着适应一下关于Composer关于Xcom和Jinja模板的2.2气流基础知识。我的BashOperator代码如下(它使用文件名中的执行日期)。此文件名作为Env变量传递，以便将{{ ts }}识别为模板(如果我将它作为Params变量传递，则不会解析执行日期，因为Params没有根据模板化)：
writeToFile1Bash = BashOperator(key="file1name", value=$file1name) }}"'
浏览 1提问于2022-05-17得票数 0
回答已采纳
我对Airflow和Python是个新手。我想要做的是从Bash命令中获取结果，并使用该返回组成连接字符串： import pyodbc as odbc import datetimeas dt from airflow.decorators import dag, taskwsl_ip",
浏览 27提问于2021-11-08得票数 0
回答已采纳
我尝试通过airflow cli test命令测试两个任务`当我通过airflowcli test命令运行第二个任务时，我只是得到了None作为返回值，但正如我在这里读到的：它应该工作，至少xcom_push是明显工作的，为什么不是xcom_pull？示例代码：
t1 = BashOperator</em
浏览 16提问于2019-11-15得票数 0
云服务器
ICP备案
对象存储
云直播
实时音视频
腾讯云开发者
