同时运行多个实例的Airflow on demand DAG
Airflow on demand DAG是指使用Airflow工具来管理和调度数据处理任务的一种方式。它允许用户根据需要动态地创建和运行多个实例，以满足不同的业务需求。
Airflow是一个开源的数据管道编排工具，它提供了一种可编程的方式来定义、调度和监控数据处理任务。它使用DAG（有向无环图）来表示任务之间的依赖关系，并提供了丰富的操作符和插件来支持各种数据处理操作。
同时运行多个实例的Airflow on demand DAG具有以下优势：
- 灵活性：通过动态创建和运行多个实例，可以根据实际需求灵活地调整任务的并发度和资源分配，以提高系统的灵活性和可扩展性。
- 资源优化：通过根据实际需求动态创建和销毁实例，可以有效地利用资源，避免资源的浪费和闲置。
- 高可靠性：Airflow提供了任务调度和监控的功能，可以确保任务按照预定的顺序和时间执行，并提供了任务失败重试和告警机制，以提高任务的可靠性和稳定性。
- 可视化界面：Airflow提供了直观的Web界面，可以方便地查看和管理任务的状态、依赖关系和执行历史，提高了任务管理的效率和可视化程度。
Airflow on demand DAG适用于以下场景：
- 数据处理和ETL：通过Airflow on demand DAG可以方便地定义和管理数据处理任务，包括数据抽取、转换和加载等操作，适用于各种数据处理和ETL场景。
- 批量任务调度：通过Airflow on demand DAG可以灵活地调度和管理各种批量任务，如数据分析、报表生成、定时任务等。
- 实时数据处理：通过Airflow on demand DAG可以结合流处理框架，如Apache Kafka、Apache Flink等，实现实时数据处理和流水线任务的调度和管理。
腾讯云提供了一系列与Airflow相关的产品和服务，包括：
- 腾讯云容器服务（Tencent Kubernetes Engine，TKE）：提供了高度可扩展的容器化环境，可以方便地部署和管理Airflow容器。
- 腾讯云函数计算（Tencent Cloud Function）：提供了无服务器的计算服务，可以根据实际需求动态地创建和运行Airflow任务。
- 腾讯云消息队列（Tencent Cloud Message Queue，CMQ）：提供了可靠的消息队列服务，可以与Airflow结合实现任务的异步调度和消息通信。
- 腾讯云数据库（TencentDB）：提供了可靠的云数据库服务，可以作为Airflow的元数据库存储任务的元数据和状态信息。
更多关于腾讯云相关产品和服务的详细介绍，请参考腾讯云官方网站：腾讯云。
相关·内容
我正在尝试运行一个Airflow dag，它创建一个EMR集群，添加一些步骤，检查它们，最后终止创建的EMR集群。但是当我运行Airflow Dag时，它一直处于运行状态，没有显示任何错误或日志。 有人能告诉我我哪里做错了吗？是否有我应该添加的缺少的参数？或者是dag时间表的问题？ import airflow
from airflow import DAG
from airflow.contrib.operators.emr_create_job_flow_operator import
EmrCreateJobFlowOperator
from airflow.contrib.oper
浏览 16提问于2019-02-04得票数 3
3回答
在我的一些Apache Airflow安装中，计划运行的DAG或任务即使在调度程序似乎没有完全加载时也不会运行。如何增加可以并发运行的DAG或任务的数量？
类似地，如果我的安装处于高负载下，并且我想限制我的Airflow工作人员拉取排队任务的速度(例如减少资源消耗)，我可以调整什么来降低平均负载？
浏览 2提问于2019-05-30得票数 45
回答已采纳
我正在与virtualenv合作。我正在尝试使用DAG文件夹中的包。airflow_home目录的当前状态为：
airflow_home/airflow.cfg
airflow_home/airflow.db
airflow_home/dags/__init__.py
airflow_home/dags/hello_world.py
airflow_home/dags/support/inner.py
airflow_home/dags/support/__init__.py
hello_world.py有代码：
from datetime import datetime
from air
浏览 1提问于2017-08-20得票数 15
要求DAG一个接一个地运行，并且每个DAG都成功运行。
我有一个主DAG，在这个主DAG中，我调用所有DAG来依次执行
在每个dag_A、dag_B、dag_C中，我都必须给schedule_interval = None，并在GUI中手动打开
我正在使用ExternalTaskSensor，因为即使在第一个dag_A中的所有任务都完成之前，它就启动了第二个dag_B，为了避免这样的问题，我正在使用ExternalTaskSensor.If更好的实现，请告诉我。
不知道我在这里错过了什么
代码: master_dag.py
import datetime
import os
from date
浏览 5提问于2020-06-25得票数 3
1回答
我刚刚开始使用气流，我基本上想运行我的dag来加载历史数据。所以我在运行这个命令
airflow backfill my_dag -s 2018-07-30 -e 2018-08-01
而气流只在2018-07-30年间运行我的雷达。我的预期是2018-07-30,2018-07-31和2018-08-01的气流。下面是我的dag代码的一部分：
import airflow
import configparser
import os
from airflow import DAG
from airflow.contrib.operators.databricks_operator impor
浏览 4提问于2020-02-26得票数 1
回答已采纳
execution_date_gte参数似乎对下面的气流REST dagRuns调用没有任何影响。
curl -X GET 'http://localhost:8080/api/v1/dags/demand_forecast/dagRuns' -H 'Cache-Control: no-cache' -H 'Content-Type: application/json' -d '{"execution_date_gte": "2023-02-02T00:00:00+00:00"}' --user &
浏览 3提问于2022-09-07得票数 0
回答已采纳
我在运行气流的教程。tutorial.py中的内容如下：
"""
Code that goes along with the Airflow located at:
http://airflow.readthedocs.org/en/latest/tutorial.html
"""
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
default_ar
浏览 1提问于2018-09-04得票数 0
1回答
我的Airflow per服务器已启动并运行，因为其他作业正在按计划运行。我添加了一个新的DAG，每5分钟执行一次。添加后，我第一次手动运行它，它完成了。然而，在那之后，它不会每隔5分钟再次运行一次。 Dag代码如下 from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
current_date = datetime.now()
default_args = {
"owner":
浏览 12提问于2019-06-11得票数 0
我正在使用Airflow EMR操作符创建一个AWS EMR集群，该集群运行包含在S3中的Jar文件，然后将输出写回S3。它似乎能够使用S3中的Jar文件运行作业，但我无法让它将输出写入S3。当作为AWS EMR CLI Bash命令运行它时，我能够让它将输出写入S3，但我需要使用Airflow EMR操作符来执行此操作。我在Airflow步骤配置和Jar文件的环境配置中都设置了S3输出目录，但仍然无法让操作员对其进行写入。 这是我的Airflow DAG的代码 from datetime import datetime, timedelta
import airflow
from airf
浏览 13提问于2019-09-12得票数 0
回答已采纳
当您遵循本教程时：
或者当您创建任何DAG并尝试使用此脚本使用API运行时：
for i in {1..10}; do
RUN_ID=BASH_REST_CALL__$(uuidgen)
curl -X POST \
http://localhost:8080/api/experimental/dags/my_test_dag/dag_runs \
-H 'Cache-Control: no-cache' \
-H 'Content-Type: application/json' -d "{\
浏览 2提问于2019-01-29得票数 2
我有多个dag使用芹菜执行程序，但我想使用Kubernetes Executor运行一个特定的dag。我无法推断出一个好的和可靠的方法来实现这一点。
我有一个声明要使用CeleryExecutor的airflow.cfg。我不想改变它，因为除了一个dags之外，所有的dags都需要它。
# The executor class that airflow should use. Choices include
# SequentialExecutor, LocalExecutor, CeleryExecutor
executor = CeleryExecutor
我的dag代码：
from da
浏览 2提问于2019-05-27得票数 6
4回答
我在不同的AWS机器上运行一个气流服务器和工作人员。我同步了它们之间的dags文件夹，在这两个文件夹上运行了airflow initdb，并在运行airflow list_tasks <dag_id>时检查了dag_id是否相同
当我运行调度程序和worker时，我在worker上得到以下错误：
airflow.exceptions.AirflowException:找不到dag_id：。要么该守护进程不存在，要么它无法解析。..。命令.-本地/home/ubuntu/airflow/dags/airflow_tutorial.py‘-sd
问题在于，由于正确的路径是(/h
浏览 4提问于2017-04-05得票数 17
我正在和BigQueryOperator一起试验气流。我想稍后我会使用，但我希望它首先在本地运行。我已经启动了气流，运行了一个BashOperator，我也可以运行airflow test <dag> <task>，其中task是我想要运行的大型查询任务，但是当我从UI触发DAG时，bigquery任务永远不会排队。相反，他们拥有REMOVED状态，什么都不会发生。
我的DAG定义如下：
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from dateti
浏览 0提问于2018-05-04得票数 11
我刚刚在Python3和Composer image版本composer-1.4.0-airflow-1.10.0上设置了一个Cloud Composer Environment。否则，所有设置都是“常用的”；即没有配置覆盖。
我正在尝试测试一个非常简单的DAG。它在我的本地Airflow服务器上运行没有问题，但在Cloud Composer上，web服务器的任务信息视图显示消息Dependencies Blocking Task From Getting Scheduled
依赖关系是Unknown，原因如下：
All dependencies are met but the task in
浏览 19提问于2018-12-20得票数 3
我想知道airflow backfill到底是如何工作的，因为它可以帮助解释我在气流方面遇到的一些问题。它是否只是启动了某种调度器进程，而不考虑其余的DAG，并覆盖了start_date和end_date？或者不仅仅是这样？ 我的用例:有时我想为一个DAG重新运行一周，通常是当我向其添加一些新功能时。为此，我使用CLI (airflow clear)清除所需日期的任务实例。但有时，我可能会改变主意，认为这周不是重新运行的合适时机。因此，我暂停了DAG，并使用命令airflow backfill -m将清除的任务实例设置为成功。 当我这样做的时候，相关的任务实例会处于关于catchup=Tru
浏览 16提问于2019-06-22得票数 1
回答已采纳
我有一个DAG，其中有多个任务排成了简单直接的依赖关系。
import datetime as dt
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.settings import log
def task1_cb(ds, **kwargs):
log.info('Task1 Complete for date: %s' % kwargs.get('end_date'))
def task2_cb
浏览 0提问于2018-12-18得票数 1
我正在使用Local Executor。我有一种情况，我为每个请求id生成了唯一的dags，例如1.py，2.py。
1.py假设有两个任务，2.py有3个任务。我也会定期收到更多的dags，比如3.py,4.py等。
是否存在为每个新ID /请求ID创建dag的问题。
我观察到Scheduler一直在给出这个日志。
Started a process (PID: 92186) to generate tasks for /Users/nshar141/airflow/dags/3.py - logging into /Users/nshar141/airflow/logs/scheduler
浏览 1提问于2018-05-08得票数 2
我对气流很陌生。我想我已经阅读了气流文档中关于调度的所有文章，但我似乎仍然不能让我的DAG运行在start_date+schedule_interval之后(即没有任务实例)。我用的是码头。我想知道我缺少了一个调度Dags的命令，尽管我在使用教程代码时并不是这样。
这是我的文件。
FROM ubuntu:latest
FROM python:3
RUN apt-get update -y
RUN apt-get install -y python-pip python-dev apt-utils build-essential
RUN pip install --upgrade pip
#
浏览 1提问于2017-03-10得票数 0
2回答
我目前正在尝试建立一个数据管道使用puckel对接-气流1.10.9图像在窗口。
当函数从DAG文件中被调用到DAG时，它工作得很好，我可以访问代码的输出。
但是，当我试图在与my_scripts目录相同的级别从my_dags目录导入函数时，我会得到以下错误：
NameError: name 'reorder_columns' is not defined
下面是用于气流的文件夹结构：
airflow
|
dags
├── dockerfiles
| ├── Dockerfile
|
├── docker-compose.yml
|
├── trading_pipeli
浏览 12提问于2022-05-11得票数 0
我们希望在Dag中的Dag触发器期间从UI读取cli输入传递给dag。我尝试了下面的代码，但它不起作用。在这里，我将输入传递为{“kpi”：“ID123”}，并希望在函数get_data_from_bq中打印此ip值。
from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.python_operator import PythonOperator
from airflow import models
from airflow.models import Variabl
浏览 1提问于2021-12-30得票数 0
1回答
我需要把DAG的运行时间改为中午。我遵循了来自不同问题的建议，删除了DAG，并上传了一个带有新dag_id的重命名版本。尽管最初的DAG已从dag_1.py重命名为new_dag_1.py，但当我上传这个新的DAG版本时，dag_id更改了，日程间隔从'@daily‘更改为’012** *‘当我上传这个新的dag版本时，这个dag主页的计划部分仍然是'1天，0:00:00’。
我用这些参数上传了一个完全不同的DAG，调度间隔正确地显示了‘012** *’，所以更改的DAG中肯定有什么东西，或者是一些元数据，阻止气流将它视为新的。除了我更改的dag文件的名称和更改的dag_id之
浏览 19提问于2022-07-17得票数 0
回答已采纳
我知道如何使用变量动态地运行DAG的任务，在为同一个DAG触发多次运行之前，它运行得很好。
也就是说，一旦在data/to/load/ dir下创建了一个包含文件的新目录，我在某个地方编写了一个脚本，它将触发airflow variables -set dir data/to/load/$newDir，然后是airflow trigger_dag dyn_test。现在，假设在data/to/load/下创建目录"a“和"b”(类似的时间)，这将使airflow variable + airflow trigger_dag调用两次，变量集调用上有两个不同的输入(一个后缀为
浏览 4提问于2019-04-14得票数 1
1回答
我们正在简化airflow代码库的构建/部署管道。
是否有人有使用CI/CD工具为apache airflow构建和部署管道的经验？
你如何在不同的环境中部署你的airflow代码库，包括DAG/插件/操作符，比如测试，试运行，生产等。
您如何管理每个环境的airflow.cfg配置？
您在何处管理每个环境的配置。
浏览 0提问于2018-08-31得票数 6
1回答
几个小时前，这段代码运行良好，但突然之间，我的伙伴们开始陷入“排队”状态。
下面是我试图运行的内容(我只是手动触发)：
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime as dt
def test_function():
print('Hello there')
default_args = {
'owner': 'airflow',
&#
浏览 3提问于2022-05-24得票数 0
我在Airflow 1.10上有一个奇怪的bug。我想试着发送电子邮件和微软团队的通知。我做了一个小的哑巴DAG来试一下。一切运行正常，但我连续收到两个通知。团队中的2封电子邮件和2条消息。 我在团队中使用了这个：https://github.com/mendhak/Airflow-MS-Teams-Operator 这里是Dags： from datetime import datetime
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.ope
浏览 34提问于2020-06-17得票数 4
1回答
我正在尝试在本地测试DAG，并在Docker中在本地设置气流。
当我在localhost上使用web登录时，我可以看到所有的示例dags，但是我试图测试的一个dag出现在顶部，有以下错误：
Broken DAG: Traceback
....
from airflow.providers.snowflake.transfers.s3_to_snowflake import S3ToSnowflakeOperator
ModuleNotFoundError: No module named 'airflow.providers.snowflake'
我已经在当地安装了气流
pip
浏览 12提问于2021-12-22得票数 1
2回答
我有以下气流代码：
dag = DAG(
dag_id='example_python_operator', default_args=args,
schedule_interval=None)
for i in range(5):
t = BashOperator(
task_id = 'some_id',
params={'i':i},
bash_command = "python airflow/scripts/a.py '{{ execution_da
浏览 0提问于2018-06-12得票数 0
回答已采纳
在气流网络服务器UI中无法打开DAG时出现问题。
有一点需要注意的是，DAG最初是在尝试运行时导致超时错误，所以我已经编辑了airflow.cfg文件，使其具有行.
dagbag_import_timeout = 300
在做了这个改变之后，跑..。
airflow list_dags
可以看到该守护进程成功构建。
然后转到webserver，刷新UI中的dag，将DAG状态切换到" on "，单击DAG尝试查看图形视图。
要么得到关于超时的消息就像..。
中断DAG: /home/气流/气流/dags/mydag.py超时，PID: 44818
(尽管dag在airflow
浏览 5提问于2020-01-16得票数 0
回答已采纳
1回答
我正在尝试使用druid hook将数据从hdfs加载到druid，下面是我的dag脚本：
from datetime import datetime, timedelta
import json
from airflow.hooks import HttpHook, DruidHook
from airflow.operators import PythonOperator
from airflow.models import DAG
def check_druid_con():
dr_hook = DruidHook(druid_ingest_conn_id='DRUID_IN
浏览 1提问于2017-08-10得票数 0
2回答
我正在创建一个具有可配置数量的执行器的动态dag。
dag中给出的“并发”参数不一致，因为这并不总是产生那么多并行任务。
将airflow.cfg设置为：
parallelism = 32
dag_concurrency = 16
任何想法，为什么这可能不会像预期的那样工作，或任何其他设置，我需要改变。请提个建议。
谢谢
浏览 0提问于2018-10-01得票数 0
我试着在每个月2号上午11点运行Airflow，但我做不到。我的设置是：
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': today_date,
'email': ['mymail'],
'email_on_failure': True,
'email_on_retry': True,
'retrie
浏览 3提问于2017-11-02得票数 1
2回答
尝试将Airflow进程拆分到2台服务器上。服务器A已经在独立模式下运行，其中包含DAG，我想将其设置为具有附加服务器的新设置中的worker。
服务器B是新的服务器，它将在MySQL上托管元数据数据库。
我可以让服务器A运行LocalExecutor吗，还是必须使用CeleryExecutor？airflow scheduler是否必须在具有正确DAG的服务器上运行？或者它必须在集群中的每台服务器上运行？不清楚进程之间有什么依赖关系
浏览 0提问于2017-07-12得票数 6
我有一个DAG，它有30个(或更多)动态创建的并行任务。
我在该DAG上设置了concurrency选项，以便在追赶历史记录时只运行单个DAG。当我在我的服务器上运行它时，实际上只有16个任务并行运行，而剩下的14个任务只是等待排队。
应该修改哪个设置，以便只运行一个DAG，但是所有的30+任务都并行运行？
根据的说法，它似乎是dag_concurrency或max_active_runs_per_dag之一，但前者似乎已经被concurrency设置过度驱动，而后者似乎没有效果(或者我实际上搞砸了我的设置)。下面是示例代码：
import datetime as dt
import logg
浏览 0提问于2019-03-21得票数 3
回答已采纳
1回答
这是我在dags文件夹中的数据文件。
Code that goes along with the Airflow located at:
http://airflow.readthedocs.org/en/latest/tutorial.html
"""
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from datetime im
浏览 0提问于2020-06-11得票数 0
回答已采纳
我编写了一个气流插件，它只包含一个自定义操作符(用于支持BigQuery中的CMEK )。我可以用一个任务创建一个简单的DAG，它使用这个操作符并执行得很好。
但是，如果我尝试在DAG中创建一个依赖项，从一个DummyOperator任务到我的自定义操作符任务，DAG无法在UI中加载并抛出以下错误，我不知道为什么要抛出这个错误？
中断的DAG: /home/airflow/gcs/dags/js_bq_custom_plugin_v2.py关系只能在操作符之间设置；接收到的BQCMEKOperator
到目前为止，我已经在作曲家-1.4.2-气流-1.9.0，作曲家-1.4.2-气流-
浏览 1提问于2019-01-23得票数 7
回答已采纳
我们最近从风流1.10.12升级到了气流2.1.0，并且在每个dags的UI中都会出现这样的问题错误。
我们已将版本从1.10.12升级到1.10.15，并取得了成功。然后，我们从1.10.15升级到2.1.0，并面临着这个问题。
File "/home/airflow/.local/lib/python3.6/site-packages/airflow/models/dagbag.py", line 249, in _add_dag_from_db
raise SerializedDagNotFound(f"DAG '{dag_id}' not
浏览 3提问于2022-05-10得票数 0
对于如何在多租户Airflow 2.0实例上进行秘密管理，有什么建议吗？我们知道有一个替代的后端，它将在envt变量和multistore之前使用，通过airflow.cfg配置。但是，我们如何在多租户envt中确保安全性，例如，我们如何限制用户及其DAG对秘密/连接的访问。是否可以使用访问控制来限制这一点？
我们正在设想将连接is放入DAG中，据我们所知，任何了解它们的人都可以访问连接并提取秘密，只要他能够创建自己的DAG。我们如何防止这种情况发生？
浏览 0提问于2021-07-26得票数 0
我正在使用我的DAG中的EMR CreateJobFlow、AddSteps、StepSensor和TerminateJobFlow操作符来启动EMR集群，添加步骤(2个spark app和dist-cp)，并在所有步骤完成或1个步骤失败时终止。当我有一个2步的DAG (第一步是Spark应用程序，第二步是dist-cp)时，我能够做到这一点，但是，当我有两个spark应用程序时，集群成功地运行了第一步，并且在没有移动到第二步和第三步的情况下终止。
通过一些挖掘，我可以看到气流“戳”着脚步，看看它们是否还在运行。在这种情况下，它似乎认为只有当一个步骤完成时它才是“成功的”。
我的spark应用
浏览 0提问于2018-10-29得票数 2
似乎之前已经有过关于这方面的讨论。
当我将dag部署为在特定时间运行时(例如，每天上午9点运行一次)，Airflow会在部署时立即运行dag。
dag = DAG(
'My Dag',
default_args=default_args,
schedule_interval='00 09 * * *',
start_date = datetime(2021, 1, 1),
catchup=False # dont run previous and backfill; run only latest
)
这是因为在catchup=False中，schedule
浏览 35提问于2021-10-13得票数 2
1回答
我试图在测试环境中使用多个任务来测试一个守护进程。我能够测试与dag关联的单个任务，但我希望在dag中创建多个任务，并启动第一个任务。用于测试我正在使用的dag中的一个任务。
task1.run()
正在被处决。但是，当我在后台的下游有一个接一个的任务时，同样的情况就不起作用了。
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
default_args = {
'owner'
浏览 0提问于2018-04-24得票数 13
我从元数据表中获取类别，并使用python脚本为每个类别创建动态dags。目前，我们大约有15个类别，所以每个类别都有自己的dag。我的Dag文件有3个任务，它是按顺序运行的。 使用LocalExecutor.All并行触发15个dag(dag-runs)。我们没有足够的资源(任务繁重)来并行运行所有15个dags。 有没有办法确定dag- run的优先级?5个dag应该先运行，然后再运行5个，依此类推。作业应该基于可用的资源运行，其他作业应该在queue.This中运行，应该是动态的。 有什么最好的方法来解决这个问题吗？请帮帮忙。 示例dag： from airflow import DA
浏览 55提问于2019-06-03得票数 0
回答已采纳
我正在运行空气流量2.1.4使用码头-合成和芹菜执行器。到目前为止，我已经能够从芹菜工作容器启动和运行简单的DockerOperator任务，但是现在当我尝试将一个目录从共享驱动器挂载到任务容器时，我得到了一个错误(下面的日志文件)。如果我不定义挂载参数，Dag工作得很好。因此，我猜有些信息或特权不会传递给由芹菜工人容器创建的容器。有什么建议吗?接下来要尝试什么？
DAG档案：
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetim
浏览 10提问于2021-11-03得票数 0
回答已采纳
我运行了以下测试命令：
airflow test events {task_name_redacted} 2018-12-12
...and得到了以下输出：
Dependencies not met for <TaskInstance: events.{redacted} 2018-12-12T00:00:00+00:00 [None]>, dependency 'Task Instance Slots Available' FAILED: The maximum number of running tasks (16) for this task's DA
浏览 0提问于2019-01-17得票数 4
3回答
我刚刚开始使用Airbnb的，我还不清楚回填是如何/什么时候完成的。
具体来说，有两个用例让我感到困惑：
如果我运行airflow scheduler几分钟，停止它一分钟，然后再次重新启动它，我的DAG似乎在前30秒左右运行额外的任务，然后继续正常运行(每10秒运行一次)。这些额外的任务是不是“回填”的任务在以前的运行中无法完成？如果是这样的话，我如何告诉气流不回填这些任务？
如果我运行airflow scheduler几分钟，然后运行airflow clear MY_tutorial，然后重新启动airflow scheduler，它似乎会运行大量额外的任务。这些任务是否也以某种方式
浏览 6提问于2016-10-05得票数 46
回答已采纳
1回答
我搞不懂它是如何工作的气流运行两个任务并行。
这是我的达格
import datetime as dt
from airflow import DAG
import os
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator, BranchPythonOperator
from airflow.contrib.sensors.file_sensor import FileSensor
from airflow.op
浏览 2提问于2020-03-09得票数 4
回答已采纳
当尝试通过airflow trigger_dag test_dag运行dag时，我得到了错误：airflow.exceptions.DagNotFound: Dag id test_task not found in DagModel。
运行airflow list_dags时，DAG正确列出。我还进行了检查，以确保将$AIRFLOW_HOME目录正确设置为dag所在的位置。我能让它工作的唯一方法就是运行一个特定的任务，比如airflow test test_dag test_task。运行python dags/test_dag.py时未显示任何错误。
导入后，dag文件本身中的代码：
de
浏览 8提问于2019-09-12得票数 3
作为一个气流新手，我在看example_branch_operator
"""Example DAG demonstrating the usage of the BranchPythonOperator."""
import random
from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import BranchPythonOperator
from airflow.utils.da
浏览 1提问于2021-05-19得票数 0
回答已采纳
我试图使用airflow trigger_dag dag_id来触发我的dag，但它只显示我的运行状态，并且不再执行此操作。
我搜索了很多问题，但所有的人都说我暂停了。问题是我的dag没有暂停，但也保持了运行状态。
注意:我可以在Web UI中使用一个dag来触发另一个dag。但它在命令行中不起作用。请看下面的快照
浏览 1提问于2019-03-14得票数 1
1回答
我有一个在成功和失败事件中执行动作和发送通知电子邮件的airflow作业，下面是我使用的代码。
#from builtins import range
from datetime import timedelta
from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago
from airflow.operators.email_operator import EmailOperator
from
浏览 2提问于2020-10-24得票数 1
1回答
我正在尝试创建一个DAG，它基于存储中的JSON文件动态地生成任务。我一步一步地遵循这个指南：
但是DAG会被以下消息卡住：
是否可以读取外部文件并使用它在Composer中动态创建任务？当我只从一个气流变量读取数据时，我可以这样做，但是当我读取一个外部文件时，守护进程会陷入isn't available in the web server's DagBag object状态。我需要从外部文件中读取，因为JSON的内容在每次执行时都会发生变化。
我正在使用composer-1.8.2-airflow-1.10.2。
我读到了一个类似问题的答案：
但是，我并不试图根
浏览 1提问于2020-01-10得票数 1
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频
腾讯云开发者
