当使用ExternalTaskSensor进行DAG任务依赖时，Airflow 1.10.2ETL运行缓慢？
在使用ExternalTaskSensor进行DAG任务依赖时，如果在Airflow 1.10.2版本下出现ETL（Extract, Transform, Load）运行缓慢的情况，可能是由于以下原因导致的：
- DAG任务依赖关系复杂：当DAG任务之间的依赖关系非常复杂时，ExternalTaskSensor可能需要等待所有依赖任务完成才能继续执行，从而导致运行缓慢。解决方法是优化DAG的依赖关系，尽量减少任务之间的依赖链条长度。
- ExternalTaskSensor的poke_interval参数设置过长：默认情况下，ExternalTaskSensor每隔1秒检查一次依赖任务的状态。如果将poke_interval设置过长，会导致等待任务状态更新的时间过长，从而造成ETL运行缓慢。建议将poke_interval设置为较小的值，以提高任务状态检查的频率。
- 配置中的concurrency参数设置过小：concurrency参数决定了同时执行的任务数量。如果将concurrency设置过小，可能会导致ExternalTaskSensor等待其他任务释放资源的时间过长，从而影响ETL的运行速度。建议适当增大concurrency参数的值，以提高同时执行的任务数量。
- Airflow环境配置不合理：如果Airflow的环境配置不合理，例如资源分配不足、网络延迟较大等，都可能导致ETL运行缓慢。建议根据实际情况评估和优化Airflow的配置，确保资源充足、网络通畅。
关于Airflow的更多信息和推荐的腾讯云相关产品，您可以参考以下链接：
- Airflow官方文档：Airflow Documentation
- 腾讯云的Serverless Workflow服务：Serverless Workflow
- 腾讯云的容器服务产品TKE：Tencent Kubernetes Engine (TKE)
- 腾讯云的云函数产品SCF：Serverless Cloud Function (SCF)
请注意，以上仅为示例链接，具体推荐的腾讯云产品应根据实际需求进行选择。
相关·内容
要求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.10.11中的ExternalTaskSensor来管理坐标。我开发了这个代码来测试功能：
import time
from datetime import datetime, timedelta
from pprint import pprint
from airflow import DAG
from airflow.operators.dagrun_operator import TriggerDagRunOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.o
浏览 16提问于2020-10-06得票数 5
回答已采纳
我正在尝试使用ExternalTaskSensor，它被困于戳另一个DAG的任务，这个任务已经成功地完成了。
在这里，第一个DAG "a“完成它的任务，然后通过ExternalTaskSensor触发第二个DAG "b”。相反，它被困在a.first_task上。
第一次DAG：
import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
dag = DAG(
dag_id='a',
default_ar
浏览 8提问于2017-10-18得票数 21
回答已采纳
3回答
我希望一个进程在完成另一个进程之后启动。一个解决方案是使用外部传感器功能，下面您可以找到我的解决方案。我遇到的问题是依赖的守护进程陷入了戳，我检查了这个并确保这两个dag按照相同的时间表运行，我的简化代码如下所示:任何帮助都将不胜感激。领袖达格：
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
default_args = {
'owner': 'airflow
浏览 0提问于2018-10-13得票数 9
回答已采纳
1回答
由于某些原因，宏'{{ ds }}‘不递增，执行日期递增，但它不适用于宏出于某种原因，我尝试了一个干净的新dag id，但仍然没有帮助。你知道吗？
from airflow import DAG
from datetime import datetime,timedelta
from airflow.operators.postgres_operator import PostgresOperator
from airflow.sensors.external_task_sensor import ExternalTaskSensor
import sys
sys.path.inser
浏览 1提问于2020-04-20得票数 0
我正在使用来调度批处理作业。我有一个每晚运行的DAG (A)和另一个每月运行一次的DAG (B)。B依赖于A已成功完成。但是，B需要很长时间才能运行，因此我希望将其放在单独的DAG中，以便更好地报告SLA。
如何使运行DAG B依赖于在同一天成功运行DAG A？
浏览 3提问于2016-06-25得票数 45
回答已采纳
同事们我们需要帮助。有两个dags --父级和子级，父级有自己的调度，假设'30 *** '，子'1 -17***-5‘，子节点等待父级执行，例如40分钟，如果父级以错误结束，那么子类也会崩溃，否则将执行子类的下一个任务。问题是，即使在最简单的情况下，这也不起作用，我不知道如何同步它们。我写了这样的代码：
Dag亲本
import time
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_op
浏览 5提问于2021-03-06得票数 4
在Airflow中，我们使用xcoms来表示一个DAG已经完成，另一个可以开始运行。例如，DAG 1在上午12点运行，并在完成时发送xcom。DAG 2从DAG 1读取xcom。 使用这种方法的一个限制是，根据文档，xcom只能从执行日期较早的DAG中读取，但不能从未来执行日期的DAG中读取。https://airflow.apache.org/docs/stable/_modules/airflow/models/xcom.html 举个例子，假设我有4个DAG，DAG #1，#2，#3，#4，它们都在做一些事情。然后我有了DAG #5，它可以并行运行多个作业。DAG#5出于不同的原因依赖
浏览 19提问于2020-06-11得票数 0
我正在尝试实现两个DAG之间的DAG依赖，比如A和B。DAG A每小时运行一次，DAG B每15分钟运行一次。
每次DAG启动时，它都是运行的，我希望确保DAG没有处于运行状态。
如果发现DAG正在运行，那么DAG必须等待DAG完成运行。
如果DAG A没有运行，DAG B可以继续执行它的任务。
达格A：
from datetime import datetime,timedelta
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
default_args
浏览 4提问于2021-05-29得票数 4
我想知道airflow backfill到底是如何工作的，因为它可以帮助解释我在气流方面遇到的一些问题。它是否只是启动了某种调度器进程，而不考虑其余的DAG，并覆盖了start_date和end_date？或者不仅仅是这样？ 我的用例:有时我想为一个DAG重新运行一周，通常是当我向其添加一些新功能时。为此，我使用CLI (airflow clear)清除所需日期的任务实例。但有时，我可能会改变主意，认为这周不是重新运行的合适时机。因此，我暂停了DAG，并使用命令airflow backfill -m将清除的任务实例设置为成功。 当我这样做的时候，相关的任务实例会处于关于catchup=Tru
浏览 16提问于2019-06-22得票数 1
回答已采纳
在airflow中，有没有一种方法可以将depends_on_past用于整个DagRun，而不仅仅是应用于任务？
我每天都有一个DAG，周五的DagRun在第四个任务上出错了，但是周六和周日的DagRuns仍然按计划运行。使用depends_on_past = True会在相同的第4个任务上暂停DagRun，但是前3个任务仍然会运行。
我可以在DagRun DB表中看到一个state列，其中包含星期五DagRun的failed。我想要的是一种方法，配置一个DagRun不启动，如果前一个DagRun失败，不启动和运行，直到找到一个以前失败的任务。
有没有人知道这是否可能？
浏览 2提问于2017-11-28得票数 13
我想使用ExternalTaskSensor。
TaskSensor = ExternalTaskSensor(
task_id='TaskSensor',
external_dag_id='datamart_OTT_CMS_v1',
external_task_id="end",
allowed_states=None,
浏览 34提问于2018-01-18得票数 2
我有一个python Parent Job和DAG Child Job。Child Job中的任务应该在成功完成每天运行的Parent Job任务时触发。如何添加外部作业触发器？
我的代码
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.postgres_operator import PostgresOperator
from utils import FAILURE_EMAILS
yesterday = datetime.combine(datetime.tod
浏览 7提问于2020-04-30得票数 22
回答已采纳
3回答
我需要有几个相同的(仅在参数上不同的) DAG**s**，顶级也可以与以下约束/假设一起触发：
单个顶级DAG将具有schedule_interval=None，因为它们只需要偶尔手动触发。
然而，这一系列的DAG需要每天运行。
系列中DAG的顺序和数量是固定的(在编写代码之前就知道了)，并且很少更改(几个月一次)。
无论DAG失败还是成功，触发链都不能中断。
目前，它们必须串联运行；将来可能需要并行触发。
因此，我在我的dags目录中为每个DAG创建了一个文件，现在我必须将它们连接起来，以便进行顺序执行。我已经确定了两种可以这样做的方法：
SubDagOpe
浏览 2提问于2018-07-13得票数 16
回答已采纳
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
我是第一次接触Airflow，遇到这个问题- 我在两个单独的文件中有两个DAG，其中第二个DAG应该在第一个DAG完成后运行。为此，我使用ExternalTaskSensor检查第一个DAG是否已运行。 DAG -1： spark_dag = DAG(
'SPARK_DAG',
default_args=default_dag_args,
schedule_interval=timedelta(days=1)
)
submit_spark_PROG_ACTN = BashOperator(
task_id='S
浏览 19提问于2020-08-03得票数 1
回答已采纳
在我的本地机器上，我创建了一个虚拟环境并安装了Airflow。当dag或plugin需要python库时，我会通过pip将其安装到相同的virtualenv中。
如何跟踪哪些库属于dag，哪些库用于airflow本身？我最近删除了一个dag，并想移除它正在使用的库。这相当耗时，我祈祷我没有删除正在被另一个dag使用的东西！
浏览 5提问于2019-07-04得票数 9
我已经创建了两个dags，我想手动触发schedule_interval=None。 首先，我触发"External_test_A“，它应该在"External_test_B”不被触发之前一直处于挂起状态。过了一段时间(2分钟后)，我触发了"External_test_B“DAG，它只运行一个任务first_task。 当task：first_task为success时，来自"External_test_A“的Poking for External_test_B.first_task应该完成，返回success并运行来自"External_test_A
浏览 280提问于2021-01-22得票数 1
我有dagA (早上5点)和dagB (早上6点)。这两种方法都从某个地方摄取数据并将其转储到数据仓库中。现在，我希望dagC ( ETL作业)能够等待dagA和dagB的完成。
我使用的是ExternalTaskSensor而不是TriggerDagRunOperator，因为我不认为摄入层会触发下游的任何东西。我读过 ，声明我应该同时运行dags。
现在，这部分让我感到困惑，因为如果我要遵循这一点，这是否意味着我所有的气流工作都会同时开始，下游的工作会一直持续，直到上游准备就绪？这是否也意味着dagA和dagB必须同时启动，即使它们之间没有依赖关系？
dagA = DAG('dag
浏览 0提问于2019-08-02得票数 0
回答已采纳
我使用root帐户在集群上安装了Airflow。我知道这是不好的练习，但它只是测试环境。我创建了一个简单的DAG：
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
dag = DAG('create_directory', description='simple create directory workflow', start_date=datetime(2017,
浏览 9提问于2017-10-30得票数 3
回答已采纳
帮我破解这个。我正在寻找一个优雅的解决方案，可以在具有独特的ExternalTaskSensor函数的气流中动态生成execution_date_fn任务，同时避免函数作用域产生的问题。
我正在尝试创建一个DAG，它依赖于其他几个不应该同时运行的DAG。我使用的是ExternalTaskSensor操作符，效果最好：
传感器任务(如sense_dag_1 )依赖于DagRun对象来查找最新的execution_date (例如dag_1 )。这个方法，在我的代码中称为get_execution_date，是通过execution_date_fn kwarg创建并传递给External
浏览 14提问于2021-01-11得票数 0
回答已采纳
1回答
我已经配置了一个dag，如果当前实例失败，下一个实例将不会运行。然而，问题是。
问题，假设任务的过去实例失败，而当前实例处于等待状态。一旦解决了这个问题，如何运行当前实例而不使过去的运行成功。我希望看到任务(Dag)失败时的历史记录。
DAG
dag = DAG(
dag_id='test_airflow',
default_args=args,
tags=['wealth', 'python', 'ml'],
schedule_interval='5 13 * * *',
浏览 4提问于2020-09-23得票数 0
1回答
我正在Ubuntu20.04中运行气流2.1.0，我在一个进程中检测到了一种优化的可能性，我需要帮助找出如何改进它。提前谢谢你。
我正在运行一个进程来执行两种任务：
Raw层=查询S3 bucketTrusted层=从S3原始桶中读取的数据库和记录文件，以及S3可信层桶中的记录
我需要允许每个任务组的可信任务在Raw任务结束后立即执行，但当发生这种情况时，我还需要触发下一个任务组的Raw任务。由于受信任的层不使用数据库，所以我可以通过允许它已经启动下一个原始任务来优化dag执行时间。也不能同时运行查询，所以只能一次运行一个原始任务。
实际流程：(未优化，原始任务本来可以启动时处于空闲状态)
T
浏览 7提问于2022-04-13得票数 0
我有多个依赖于首先运行initial_dag的DAG，然后我希望依赖的DAG能够一个接一个地运行。以下是我所拥有的：
dag = DAG(
dag_id=DAG_NAME,
default_args=default_args,
schedule_interval=None,
start_date=airflow.utils.dates.days_ago(1)
)
initial_dag = BashOperator(
task_id='initial_dag',
bash_command="python /home/ai
浏览 0提问于2021-06-21得票数 1
回答已采纳
我已经创建了一个dag，该dag在Airflow UI中可用，并将其打开以运行它。运行dag后，状态显示为可重试。在那之后，我去了服务器，使用了命令"Airflow scheduler“，然后dag就成功了。
在运行dag之前，调度程序已经启动并运行，我不确定为什么会发生这种情况。当我们创建一个新的dag时，我们需要运行气流调度器吗？想知道调度器是如何工作的。
谢谢
浏览 0提问于2020-05-20得票数 1
我已经这样定义了external_sensor：
external_sensor = ExternalTaskSensor(task_id='ext_sensor_task',
execution_delta=timedelta(minutes=0),
external_dag_id='book_data',
external_task_id='Dataframe_Windows_test',
dag = dag)
另一
浏览 140提问于2019-11-20得票数 1
在气流网络服务器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
回答已采纳
我有两个不同的DAG需要在不同的频率上运行。一种是dag1需要每周运行一次，dag2需要每天运行。现在，dag2应该只在dag1完成时运行，在每次运行dag1时都会运行。
我在两个不同的python模块中定义了两个DAG，如下所示。
dag1.py
PROJECT_PATH = path.abspath(path.join(path.dirname(__file__), '../..'))
with DAG('dag1',
default_args={
'owner': 'airflow',
浏览 2提问于2019-08-26得票数 0
回答已采纳
我刚加入一家新公司，我试着在使用它的时候学习气流。到目前为止，除了外部任务传感器之外，我已经掌握了大多数事情的基本知识。我有两个DAG，DAG的调度间隔为"0 6 * * *"，DAG的调度间隔为"0 7 * * *"，DAG等待DAG在继续之前完成。然而，DAG有时需要3小时才能完成，而在其他时候则需要10+时间。
我创建了一个ExternalTask传感器，如下所示，但它从不触发和超时，即使在DAG完成时。
ExternalTaskSensor(
task_id = "wait_sensor",
external_dag_id
浏览 18提问于2022-11-24得票数 2
回答已采纳
我们的airflow安装使用的是CeleryExecutor。并发性配置为 # The amount of parallelism as a setting to the executor. This defines
# the max number of task instances that should run simultaneously
# on this airflow installation
parallelism = 16
# The number of task instances allowed to run concurrently by the scheduler
浏览 182提问于2019-05-14得票数 12
回答已采纳
谁能解释一下为什么Airflow的ExternalTaskSensor在日程安排上正常工作，而不是在前端触发时工作？我尝试了几个版本，结果总是一样的。任务在up_fro_reschedule上停止。
example_external_task_marker_child和example_external_task_marker_parent
Task.log
[2020-03-08 22:25:25,071] {taskinstance.py:866} INFO - Executing <Task(ExternalTaskSensor): child_task1> on 2020-0
浏览 56提问于2020-03-09得票数 2
我来自使用Luigi的经验，在Luigi中，如果某个任务成功地生成了一个文件，并且该任务也未被修改，那么重新运行DAG将不会重新运行该任务，而是重用其先前获得的输出。
有什么方法可以获得与AirFlow相同的行为吗？
目前，如果我重新运行dag，它将重新执行所有任务，无论它们在过去是否生成了成功(且未更改)的输出。因此，基本上我需要一个任务被标记为成功，如果它的代码是没有改变的。
浏览 9提问于2021-09-13得票数 0
我正在运行多个实例。所以，max_active_runs=1不是我的解决方案。wait_for_downstream只能等待立即下游的任务。
从baseoperator.py代码，
wait_for_downstream:当设置为true时，任务X的实例将等待任务X的上一个实例的下游的任务在运行之前成功完成。如果任务X的不同实例更改了相同的资产，并且任务X的下游任务使用了该资产，那么这一点非常有用。注意，无论在哪里使用depends_on_past，都会强制使用wait_for_downstream。还要注意的是，只等待前一个任务实例的下游的任务；任何下游任务的状态都会被忽略。
如何等待所
浏览 4提问于2020-10-16得票数 2
回答已采纳
我得到了以下DAG
import logging
from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import BranchPythonOperator
from airflow.utils.dates import days_ago
from airflow.utils.task_group import TaskGroup
def select_next_branch():
if some_condition:
浏览 6提问于2022-05-20得票数 0
回答已采纳
我使用来转换一些运行在AWS上的oozie工作流。成功地获得了第一个版本，但是当我试图上传DAG时，气流抛出一个错误：
坏掉的DAG:没有名为'o2a'的模块
我尝试过部署pypi包o2a，两者都使用命令
gcloud composer environments update composer-name --update-pypi-packages-from-file requirements.txt --location location
来自谷歌云控制台。都失败了。
requirements.txt
o2a==1.0.1
这是代码
from airflow impo
浏览 0提问于2020-04-30得票数 0
回答已采纳
我正在尝试在另一个DAG (让我们称之为DAG A)中的任务上创建一个外部传感器(在DAG B中)，它以以下间隔运行：'schedule_interval'：'0 4,6,8,10,12,14,16,18,20,22 ** *‘。DAG B计划在每天凌晨2点运行。我想在DAG B中创建一个传感器任务，以检查DAG A中外部任务的4AM运行是否成功
我无法将我的DAG B重新安排为以4运行，因为DAG B中还有其他任务需要以2运行。我尝试更改window_size和window_offset参数，但不起作用。
ExternalTaskSensor方法已被覆盖，如下所示
f
浏览 0提问于2019-09-17得票数 4
我不知道这是不是应该是这样的，但让我说，我做了一个4任务，我只把3个任务放在蒸汽中，为什么第四个任务也会出现和运行？
例：
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2018, 1, 1),
'email_on_failure': False,
'email_on_retry': False,
'retries
浏览 0提问于2020-02-06得票数 1
回答已采纳
我有一个airflow服务，它目前作为for服务器和调度器的独立docker容器运行，两者都由postgres数据库支持。我让dags在两个实例之间同步，并在服务启动时适当地加载dags。但是，如果我在服务运行时将新的dag添加到dag文件夹(在两个容器上)，该dag会加载到dagbag中，但会在web gui中显示缺少的元数据。我可以在每次更新后运行"airflow initdb“，但感觉不太对劲。有没有更好的方法让调度程序和way服务器与数据库同步？
浏览 1提问于2018-08-17得票数 6
我尝试通过airflow cli test命令测试两个任务`
第一个任务运行时，auto将最后一个控制台推送到xcom，我在airflow GUI中看到了预期的值some value
当我通过airflow cli test命令运行第二个任务时，我只是得到了None作为返回值，但正如我在这里读到的：它应该工作，至少xcom_push是明显工作的，为什么不是xcom_pull？
有人给出了如何让它工作的提示吗？Provide context设置为true。
示例代码：
t1 = BashOperator(
task_id='t1',
bash_command=&#
浏览 16提问于2019-11-15得票数 0
我制作了一个非常简单的DAG，如下所示：
from datetime import datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
cleanup_command = "/home/ubuntu/airflow/dags/scripts/log_cleanup/log_cleanup.sh "
dag = DAG(
'log_cleanup',
description='DAG for deleting old logs
浏览 1提问于2018-04-02得票数 2
回答已采纳
当我在终端中运行它时，下面的代码运行正常：
data = []
def _step_one():
for i in range(10):
data.append(i)
def _step_two():
print(f'data: {data}')
_step_one()
_step_two()
它打印data: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
但是，当我在Apache Airflow上运行同样的命令时，第一次初始化时data是空的[]
from airflow import DAG
from airflow.o
浏览 9提问于2021-02-24得票数 0
回答已采纳
这是我的气流传感器的样子。
有一个分支任务，它检查一个条件，然后：
直接运行任务B，跳过任务A或
运行任务A，然后运行任务B
当跳过任务A时，在dag的下一次(将来)运行中，分支任务永远不会运行(执行在主任务处停止)，尽管默认触发规则是'none_failed‘并且仅跳过dag中没有失败的任务。
default_args = {
'owner': 'airflow',
'depends_on_past': True,
'wait_for_downstream': True,
'e
浏览 0提问于2021-07-16得票数 1
在一个进程运行期间读取一个dag定义文件的频率是多少？
拥有一个需要很长时间构建的大型守护进程(~1-3分钟)。当进程运行时，查看每个任务的日志，似乎每个任务在运行之前都在执行dag定义文件.
*** Reading local file: /home/airflow/airflow/logs/mydag/mytask/2020-01-30T04:51:34.621883+00:00/1.log
[2020-01-29 19:02:10,844] {taskinstance.py:655} INFO - Dependencies all met for <TaskInstance: my
浏览 4提问于2020-01-30得票数 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
我正试图从另一个人身上触发另一个人。我也在使用TriggerDagRunOperator。
我有以下两条线索。
Dag 1:
from datetime import datetime
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.operators.dagrun_operator import TriggerDagRunOper
浏览 4提问于2020-01-06得票数 0
似乎之前已经有过关于这方面的讨论。
当我将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
我目前正在使用Airflow来运行一个有几个任务的DAG (比如dag.py)，然后，它有一个要执行的python脚本(通过bash_operator完成)。python脚本(比如report.py)基本上是从云(s3)位置获取数据作为数据帧，执行一些转换，然后通过电子邮件将它们作为报告发送出去。 但我遇到的问题是，每当airflow扫描存储库的更改时(即每隔2分钟)，Airflow基本上都在运行这个python脚本report.py。因此，脚本每2分钟运行一次(因此电子邮件每两分钟发送一次！)。 有没有解决这个问题的办法？除了bash运算符之外，我们还能使用其他东西吗(请记住，在发送报告之
浏览 19提问于2021-01-06得票数 0
我创建了一个带有重试计数的气流任务，在运行我的气流测试时，它似乎并不是真正的重试。
def verify_pod_initialization(deployments):
raise ValueError("test")
verify_deployments = PythonOperator(
task_id='verify_pod_initialization',
retries=5,
python_callable=verify_pod_initialization,
op_kwargs={'deploym
浏览 2提问于2020-06-02得票数 0
回答已采纳
我在多节点环境中使用Hadoop和Spark。我已经安装了Airflow来自动执行多个spark任务。为了在多个节点上运行这个DAG，在Airflow中是Celery executor还是Kubernetes executor是最好的选择？
浏览 3提问于2019-07-09得票数 0
我有多个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
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
