我用airflow 2 Taskflow API创建了一个DAG： cf = collect_files() for i in range(0, max_parallel_plot_t
浏览 8提问于2021-02-26得票数 0
在使用Apache Airflow的TaskFlow API定义DAG时，有什么方法可以使用airflow.utils.edgemodifier.Label吗？但在中，似乎没有一种方法可以包含标签，因为DAG是通过对象在任务之间传递的方式来推断的。
有没有办法做到这一点？
浏览 5提问于2021-06-30得票数 0
假设我有一个简单的TaskFlow风格的DAG。from datetime import datetimefrom typing import Dict
schedule_interval='@once',def example_taskflow_a
浏览 16提问于2022-08-21得票数 0
1回答
如何从气流任务中获取xcom，并使用这些值创建其他任务。每一个例子： return ['task_2', 'task_3']
'dag_name', catchup=False, 'depends_on_past': False,
'retries&#x
浏览 9提问于2022-11-10得票数 0
回答已采纳
假设我的虚拟DAG定义如下： schedule_interval=None,def airflow_taskflow_api_dag(): def A(current_day: strlivy_task.outpu
浏览 3提问于2021-03-11得票数 3
1回答
我可以通过将来自任务的数据存储在一个临时文件中，然后在下一个任务中读取它--但是我希望仅使用气流资源来解决这个问题。from airflow.decorators import dag, task
def extract1,
def tutorial_taskflow_<
浏览 1提问于2022-01-19得票数 0
当我通过for循环生成所有运算符时，如何创建任务依赖项。示例：from airflow.operators.python_operator import PythonOperator
'from_conn': 'c.from_conn', },
浏览 2提问于2022-02-14得票数 0
回答已采纳
我研究了xcom拉和推的概念，这就是我在下面实现的，我也看到了一个用于气流的MongoHook，但不太确定如何使用它。import DAG
'start_date': airflow.utils.dates.days_ago(2),
dag = DAG
浏览 2提问于2019-09-09得票数 0
回答已采纳
我最近开始使用Apache气流，在使用了传统的创建DAG和任务的方法之后，我决定使用Taskflow API。然而，我遇到了一些问题，下面是我的问题。Taskflow：from airflow.utils.dates import days_ago
@dag(return 1 #Question1 -Ha
浏览 12提问于2022-07-20得票数 0
我使用REST API将参数传递给基于任务流的Dag。看看在这个论坛上提出的类似问题，下面似乎是访问传递的参数的常见方式。within a python callable of the PythonOperator:
context['dag_run'].conf['key'] 我尝试获取上下文字典 @dag(default_args=default_args, schedule_interval=None, start_date=days_ago(2</
浏览 29提问于2021-01-07得票数 6
1回答
我得到了这个损坏的DAG错误重复任务id File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/b
浏览 8提问于2021-05-25得票数 4
AddEMRStep 2.传感器3. SQLstep。我只想为两个环境创建它。with dag: task_id = "run_this", )
现在，我需要为多个环境创建这些标记。我正试着这样做
envs = ["stg","p
浏览 1提问于2022-02-11得票数 0
回答已采纳
我试图弄清楚如何在我正在创建的DAG中使用PythonVirtualenvOperator，在Airflow 2.0.1中使用TaskFlow API。样本码from time import sleepfrom airflow.utils.dates': 'airflow
浏览 8提问于2021-04-06得票数 4
import pendulumfrom airflow.decorators import dag, taskfrom custom_operators.validate_data_operator import ValidateDataOperator
from <e
浏览 1提问于2022-05-17得票数 0
想要尝试新的任务流API，我达到了需要2个parallels任务的地步。使用Airflow v1时，我习惯于这样做 task_1 >> [task_2, task_3]
[task_2, task_3] >> task_4 对于PythonOperator，我们调用任务的方式现在不同了我怎么用TaskFlow做列表呢？
浏览 17提问于2021-04-24得票数 3
回答已采纳
我希望获得SparkSubmitOperator的状态，将其转换为API所理解的值，并在SimpleHttpOperator的有效负载中传递它，以便我可以更新DB中的作业状态。我想通过使用Taskflow API来做到这一点。我编写了下面的代码，但是当我试图加载它时，我得到了这个错误：
Broken DAG: [/opt/airflow/dags/export/inapp_clicks/export.py] T
浏览 5提问于2021-08-20得票数 1
回答已采纳
2回答
1: pd.read_csv用于目录中的文件3:干净的列名5:将结果写入数据库
我将如何组织我的文件根据气流？剧本应该是什么样的呢？
浏览 1提问于2018-06-03得票数 3
回答已采纳
我试图使用taskflow API创建多个dag，该API中有一个变量传递给它们，这些变量可以由守护进程中的任务使用。例如，我正在尝试使用以下代码from datetime import datetime
print(x)
output_input_var
浏览 5提问于2021-12-08得票数 0
回答已采纳
因此，我尝试通过父抽象类(使用TaskFlow API)抽象它的构造：from airflow.decorators import): def task_2(self, data):
"""task 2"&qu
浏览 14提问于2022-04-26得票数 0
回答已采纳
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
