在Airflow上使用PythonOperator时，如何使用Python函数的返回值/

文章/答案/技术大牛

发布

1回答

python、airflow、directed-acyclic-graphs

我正在构建一个具有多个PythonOperator节点的airflow DAG。其中一个返回一个值，稍后将用作另一个运算符的参数。但是我如何存储和访问这个返回值呢？例如:我有以下函数 return a + b return c > dsum = PythonOperatorcompare = PythonOperator(

浏览 6提问于2020-02-12得票数 0

回答已采纳

2回答

Apache airflow在AWS中使用kubernetes运行dag运算符

kubernetes、airflow、aws-fargate、amazon-eks

我正在评估Apache airflow在数据环境中的生产使用，我想知道使用airflow是否可以在自适应扩展Kubernetes集群上的自包含docker环境中运行操作员。我找到了以下运算符：KubernetesPodOperator，它似乎可以完成这项工作，但我找到的唯一示例都是在Google Cloud上。我想在AWS上运行这个，但是我还没有找到任何如何做到这

浏览 16提问于2019-02-03得票数 4

2回答

使用Xcom从PostgresOperator中推送一个值并在PythonOperator中提取它

airflow

我希望将查询“从<table_name>中选择计数(*)”的结果从PostgresOperator中推送到PythonOperator，后者是pythonCallable，该值将用于python函数。我如何在其中使用XCom？我尝试过不同的方法，但每次我在python函数中得到一个未定义的变量时都会出错。def <python_function_

浏览 5提问于2022-07-13得票数 0

3回答

如何在虚拟环境中运行Airflow PythonOperator

python、virtualenv、airflow

/python_files/python_task1.py',对于这样的东西，我如何使用from airflow.operators.bash_operator import PythonOperator dag=d

浏览 1提问于2018-04-10得票数 19

1回答

当PythonOperator尝试调用API并下载数据时，Airflow DAG失败

python、airflow、etl、sigabrt

我第一次尝试在我的笔记本电脑上配置Airflow (不使用docker，只遵循文档)。我的目标是设置一个简单的ETL作业。import DAGfrom airflow.utils.dates importprint语句传递一个伪函数时，DAG

浏览 42提问于2020-11-08得票数 0

回答已采纳

3回答

如何使用AirFlow来运行python文件的文件夹？

python、airflow

我在Python文件的文件夹中有一系列python任务: file1.py，file2.py，...Task2:执行file2.py (使用</em

浏览 19提问于2016-09-21得票数 29

回答已采纳

1回答

有没有办法在Airflow中通过XComArgs将多个参数传递给可调用的python？

python、airflow

从Apache Airflow 2.0.0版本开始，我们可以轻松地通过装饰器@task将一个函数的输出作为输入传递给另一个函数，如以下代码所示fromairflow.decorators import taskfrom airflow.operators.pythonXComArgs (@task背后<

浏览 63提问于2021-05-14得票数 1

回答已采纳

1回答

如何在气流中调度Python脚本？

python、docker-compose、airflow、airflow-scheduler

我在Windows中有以下dir/file结构： | |_dags |_file_mover.py在first_dag.py中，我有一个触发file_mover.py的脚本。所以当我在终端做docker-compose up --build检查webserver localhost:8080时，我确实看到了气流中的first_dag。

浏览 2提问于2021-06-16得票数 1

2回答

我正在创建我的第一个使用subdag的dag，并导入subdag运算符 `from airflow.operators.subdag import SubDagOperator` 然而，我一直收到流错误"Broken : /usr/local/airflow/dags/POC_Main_DAG.py没有名为‘airflow.operators.subdag’的模块“ 我还尝试导入虚拟操作符，但得到了相同

浏览 8提问于2021-01-06得票数 2

回答已采纳

1回答

使用trigger_rule的气流ONE_FAILED引起dag故障

airflow

我想要实现的是创建一个任务，如果有的话，在那里发送通知--进程下的一个任务失败了。我将触发规则应用于以下任务：task_id='Error_Buzz',bash_command='pythonnotifications/dagLevel_Notification.py') ,catchup = False batch>>batch

浏览 0提问于2018-06-21得票数 2

1回答

我可以在cloud composer DAG中执行python脚本吗？

python、directed-acyclic-graphs、google-cloud-composer

( cloud composer的第一次用户)我见过的所有示例都在DAG中定义了非常简单的python函数。我有多个冗长的python脚本要运行。我可以将这些放入任务中吗？如果是这样，那么是使用PythonOperator还是从BashOperator调用它们更好呢？, default_args = default_dag_args) as dag: do_stuff1 = python_operator.<em

浏览 17提问于2021-07-24得票数 2

回答已采纳

1回答

将python脚本转换为气流PythonOperator

python、airflow

我有一个运行在CronJob上的python脚本。我想把它转换成DAG与PythonOperator(s)，因为我们现在正在转换为气流。假设我有函数：a(),b(),c(),d()，它们的执行顺序是：a->b->c->ddef a(): print**这只是我为所有函数编写的代码更复杂的</em

浏览 0提问于2018-06-10得票数 2

回答已采纳

4回答

气流中的Python脚本调度

python、apache-spark、scheduling、reload、airflow

大家好，我需要使用气流调度我的python 文件(包含从sql和一些联接中提取的数据)。我已经成功地将气流安装到我的linux服务器上，而且我也可以使用气流的and服务器。但是，即使在阅读了文档之后，我也不清楚到底需要在哪里编写调度脚本，以及如何将该脚本应用到气流see服务器中，这样我就可以看到状态。就配置而言，我知道dag文件夹位于我的主目录中的位置，也知道示例dags位于何处

浏览 4提问于2017-01-18得票数 31

3回答

在气流DAG中导入本地模块(python脚本)

python、airflow

我正在尝试将本地模块( python脚本)导入到我的DAG中。目录结构：├── dag│ └── my_DAG.py └── subfoldera = some_function() 我在气流中出现了一个错误，上面写着“坏掉的DAG: my_DAG

浏览 0提问于2018-05-03得票数 39

1回答

在Airflow中如何在PythonOperator中登录Python函数

airflow

我在Airflow DAG中使用了PythonOperator，并且我需要在操作符的Python函数中打印一些内容。我试着打印，但显然不起作用。我不太确定这是否会奏效。接下来，我尝试在PythonOperator中传递self.log，但我不确定如何传递该引用。task = PythonOperator( python</

浏览 10提问于2018-08-26得票数 5

回答已采纳

1回答

在python中执行定义为DAG的任务的简单方法？

python、flask、airflow、google-cloud-run、directed-acyclic-graphs

我正在以一种复杂的方式运行一系列相互依赖的任务。我想将这些依赖关系描述为DAG (有向无环图)，并在需要时执行该图。我一直在关注airflow，并写了一个虚拟脚本： from datetime import datetime from airflow.operators.python：我的使用不会涉及任何类型的日程安排/开始日期/

浏览 64提问于2021-10-16得票数 0

1回答

Airflow PythonOperator任务无论如何都不会失败

python、docker、airflow

我正在尝试使用docker中的airflow 2.0.1来运行我的python项目。下面是目录结构： dags在这里: path_to_airflow/dags/ python项目代码在这里: path_to_airflow/dags/utils 我遇到了处理异常的奇怪态度:我的PythonOperatorimport DummyOperator from <

浏览 25提问于2021-09-25得票数 0

回答已采纳

1回答

通过“for”循环实例化多个任务时发生气流依赖错误

python、python-3.x、airflow

它从dash_workers.py导入函数(不包括，但是--这会有帮助吗？)并将这些函数实现为PythonOperator定义的任务。这里的应用程序是使用函数get_id_creds从SQL表中提取ID列表，然后在每个ID的基础上生成详细的数据配置文件。这两个函数都在内部使用MySqlHook，并且我已经在独立的基础上<

浏览 2提问于2017-07-21得票数 1

回答已采纳

1回答

在气流中使用jinja模板的F-字符串将动态值传递给op_kwargs

python、airflow、jinja2、keyword-argument、f-string

我试图使用xcom_pull在气流中动态地提取值。当我将op_kwargs中的f-字符串拼凑在一起时，下面提到的格式并不适用于我。如果有人能帮我的话我很感激。

浏览 12提问于2022-11-18得票数 1

1回答

如何将数据从python_callable传递到on_success_callback？

callback、airflow

我有一个带有调用函数的PythonOperator的DAG。我还有一个回调函数。我想把一些信息从可回调函数传递给回调函数。我该怎么做呢？我尝试使用上下文字典：from pprint import pprintfrom airflow.operators.python_operatorimp

浏览 9提问于2019-11-19得票数 1

点击加载更多