无法使Airflow DAG在每天的特定时间运行
Airflow是一个开源的任务调度和工作流管理平台，它允许用户定义、调度和监控工作流任务。Airflow中的任务被组织成有向无环图（DAG），其中每个节点表示一个任务，边表示任务之间的依赖关系。
要使Airflow DAG在每天的特定时间运行，可以使用Airflow提供的调度器和调度器插件来实现。具体步骤如下：
- 创建一个DAG对象：在Airflow中，首先需要创建一个DAG对象来定义工作流。可以使用Python脚本编写一个DAG对象，并在其中定义任务和任务之间的依赖关系。
- 定义任务：在DAG对象中，可以定义多个任务。每个任务由一个Operator对象表示，可以是BashOperator、PythonOperator等。可以根据具体需求选择合适的Operator。
- 定义任务之间的依赖关系：在DAG对象中，可以使用>>运算符来定义任务之间的依赖关系。例如，如果任务B依赖于任务A完成，可以使用
A >> B来定义依赖关系。 - 设置调度时间：在DAG对象中，可以使用
schedule_interval参数来设置任务的调度时间。对于每天的特定时间运行，可以使用Cron表达式来指定调度时间。例如，schedule_interval='0 8 * * *'表示每天上午8点运行。 - 配置调度器：Airflow提供了多种调度器插件，可以根据具体需求选择合适的调度器插件。可以在Airflow的配置文件中配置调度器插件，并启动调度器。
- 启动Airflow服务：在配置好DAG对象和调度器后，可以启动Airflow服务。启动后，Airflow会根据配置的调度时间自动触发DAG的运行。
推荐的腾讯云相关产品：腾讯云容器服务（Tencent Kubernetes Engine，TKE）是腾讯云提供的一种高度可扩展的容器管理服务，可用于部署、管理和扩展容器化应用程序。TKE提供了强大的容器编排和调度功能，可以方便地部署和管理Airflow的容器化应用。
更多关于腾讯云容器服务的信息，请访问：腾讯云容器服务
请注意，以上答案仅供参考，具体的实现方式和产品选择应根据实际需求和环境来确定。
相关·内容
我们一直在将cron作业转换为Airflow DAG，我很难弄清楚DAG在Airflow中的调度方式。一些day需要在一天中的特定时间(即上午7点)运行，其他day需要在一个月的特定日期/时间运行(即每月15日上午6点)。
一般来说，Airflow似乎每天都能正确运行DAG。因此，带有'start_date': datetime(2017,4,7)的schedule_interval = '0 7 * * *每天早上7点运行
然而，对于每月的DAG (schedule_interval = '0 6 15 * *'和'start_date'
浏览 2提问于2017-05-29得票数 4
当我安排DAG在每天的特定时间运行时，DAG根本不会执行。但是，当我重新启动Airflow when服务器和调度程序时，DAG在该特定日期的计划时间执行一次，并且从第二天起不再执行。我使用的是带有python 2.7.6的Airflow版本v1.7.1.3。下面是DAG代码：
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
import time
n=time.strftime("%Y,%m
浏览 2提问于2016-11-21得票数 14
我们有一个按小时运行的气流数据集，任务更新和覆盖BigQuery中的日期分区表。
在对这些表的查询和/或模式进行调整之后，我们想要回填几天的现有分区，但是回填所有运行都是巨大的浪费，因为每小时运行一次只会覆盖相同的分区24次，然后再转到第二天。
我们可以使用airflow list_dag_runs列出所有运行并筛选出每天的最后一个运行，但是是否有一种方法可以每天只回填/清除这些最后的运行，而不需要每天重新运行24个实例呢？
airflow clear和airflow backfill命令具有指定开始日期和结束日期的选项，但不指定特定的实例时间，因此它们将导致每次日期重新运行24次，它们都将执
浏览 3提问于2020-06-10得票数 0
回答已采纳
我的DAG是这样的，
dag = DAG('testing',description='Testing DAG',schedule_interval='0 4,15 * * *')
t1 = BashOperator(task_id = 'testing_task',bash_command = 'python /home/ubuntu/airflow/dags/scripts/test.py',dag=dag, start_date=datetime(2018, 2, 8))
我想将其安排为每天下午3点和凌晨4
浏览 30提问于2018-02-08得票数 1
回答已采纳
我有一个气流DAG，这是没有意义的回填。我想，使用气流1.8，你可以给DAG参数catchup=False，所以它只会开始最近的工作。也就是说，我想让DAG从午夜开始，每天运行。但。这就是现在的事情: DAG立即开始，而不是在午夜。另外，当我清除所有DAG运行时，它将立即重新启动。然后DAG将每天运行，但将在错误的时间安排它开始+1天。
我怎么能有一个DAG只开始运行最近的作业，和在特定的时间(午夜)开始？
下面是我使用的代码：
from datetime import datetime, timedelta
from airflow import DAG
from airflow.opera
浏览 3提问于2017-11-22得票数 2
回答已采纳
我有两个不同的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
回答已采纳
1回答
我有两个DAG，它们在Airflow上有不同的时间表，DAG1每小时在Redshift上创建一个表，DAG2是一个每天运行的进程，需要读取DAG1创建的表。我正面临着一个问题，如果DAG2花费的时间比平时长，那么DAG1就会开始运行，导致我在Redshift上出现并发连接问题。
是否可以在DAG1中设置传感器以检查DAG2是否正在运行，在这种情况下，是否可以等待传感器完成后再继续？
我尝试使用ExternalTaskSensor，但它只检查上一次运行是否成功，而不是检查是否正在运行。还是我错了？
有没有办法获得一个DAG的状态并使用它来重新安排另一个DAG？
浏览 5提问于2020-03-25得票数 0
我是Airflow的新手，但我已经定义了一个Dag，每天上午9点发送基本的电子邮件我的DAG如下： from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.bash_operator import BashOperator
from airflow.operators.email_operator import EmailOperator
from airflow.utils.dates import days_ago
date_log = str(datetime.to
浏览 31提问于2021-06-26得票数 2
回答已采纳
我正在使用Airflow 设置交叉依赖的DAGS。我有一个特定的使用案例，其中我的DAG B要求DAG A首先运行-但是，如果DAG A延迟足够长时间，DAG B仍应运行。因此，我本质上是在寻找一种方法来连接两个传感器之间的OR操作。
假设DAG B需要每天下午5点运行，那么我将在代码中这样做：
while True:
CURRENT_TIME = getCurrentTime()
if DAG A completed OR CURRENT_TIME > 5pm:
run DAG B
这在代码中要简单得多，但是看不到Airflow是如何做到的。
浏览 12提问于2020-09-03得票数 1
我有一个气流DAG，它在文件存在时工作得很好，但是当源文件不存在时，错误->失败。
随机地，我从给定的源接收文件，我的DAG收集并处理这些文件。虽然我需要每天运行DAG，但文件不一定每天都在那里。可能是星期一，星期三，甚至星期天晚上。
我不担心没有新文件的日子。我担心有一天新的文件来了，然后它就坏了。
我如何告诉DAG，当不存在文件时，成功地优雅地退出？我的DAG在下面(请忽略时间表设置。我仍在开发模式中)：
import airflow
from airflow import models
from airflow.operators.gcs_to_bq import GoogleCl
浏览 1提问于2019-10-09得票数 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
我知道以前有人问过这个问题，但我想把它说得很简单。我正在使用Astronomer运行Airflow v1.10.10。每个dag运行(每天)可以有不同数量的任务。此任务数将由给定dag运行(即当天)的数据库表中的行数定义。
我将在dag文件中运行一个查询，并创建一个如下列表：
list = client.query(SELECT * FROM TABLE WHERE DATE = {{ds}})
然后，我将根据列表的大小创建任务：
for i in range(1, len(list)):
sample_airflow_operator(name=i + '_operator&
浏览 2提问于2021-06-03得票数 0
我使用Airflow来管理ETL任务的执行和调度。已经创建了一个DAG，它工作正常。但是当通过cli手动触发dag时传递参数是可能的吗。
例如:我的DAG每天下午1点半运行，处理昨天的数据(时间范围从昨天下午1点半到今天1点半)。数据源可能存在一些问题。我需要重新处理这些数据(手动指定时间范围)。
因此，我是否可以创建这样一个气流DAG，当它被安排时，默认时间范围是从昨天的01:30到今天的01:30。然后，如果数据源有任何问题，我需要手动触发DAG，并手动传递时间范围作为参数。
据我所知，airflow test有可以将参数传递给任务的-tp。但这只是为了测试特定的任务。而且airflow
浏览 0提问于2018-12-07得票数 37
回答已采纳
我有dag： args = {
'owner': 'airflow',
'start_date': datetime(2020, 4, 1),
'retries': 1,
'retry_delay': timedelta(minutes=5)
}
dag = DAG(
'test',
default_args=args,
schedule_interval="@daily",
max_active_runs=1
)
浏览 45提问于2020-11-27得票数 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
我运行的是Apache Airflow 1.10.0，我想利用新的时区感知Dag功能。我必须承认，气流调度器有点混乱，我不太确定如何完成我想要做的事情。我正在尝试定义一个Dag，它将在每天午夜过5点(东部时间)运行。
到目前为止，我已经尝试使用钟摆来定义具有时区感知的start_date的Dag。我的计划间隔是timedelta(days=1)。由于某些原因，这导致在看似奇怪的时间运行12:00，等等。
我当前的Dag定义：
...
dag_tz = pendulum.timezone('US/Eastern')
default_args = {
'owne
浏览 0提问于2019-05-10得票数 1
1回答
我有一个只有一个任务的dag，我希望它每天只触发一次。我的问题是，当时机成熟时，它会被多次触发。因此，每天的任务运行4次而不是1次。我设置了许多配置来解决这个问题，包括： 'retries': 1 catchup=False, max_active_runs=1 我还增加了退休之间的时间间隔，认为可能airflow认为任务已经失败/没有开始，因为任务可能需要一些时间才能完成。 我还基于此answer将在该dag中运行的所有代码移动到utils文件夹中。 但我不知道我错过了什么。有人能帮帮忙吗？提前谢谢你。 这是dag from datetime import datetim
浏览 98提问于2021-11-09得票数 0
如何配置Airflow dag，使其在每天的指定时间执行，无论发生什么，就像crons一样。
我知道可以使用TimeSensor获得类似的行为，但在这种情况下，它取决于传感器任务，并且可能与dag执行时间冲突。
例如:使用传感器方法，如果我让传感器在0小时15分钟运行，但如果在以后执行dag，那么我的任务就会延迟，所以即使是使用传感器方法，我也需要确保Dag在正确的时间执行。
那么如何确保Dag在指定的时间执行呢？
浏览 3提问于2016-02-27得票数 18
回答已采纳
我每天晚上7:42有一个dag，然而，当我手动触发dag时，它会运行，而不是自动运行。
我正在使用官方的yaml文件和运行容器，默认情况下已经将时区设置为IST，仍然不能自动触发dags
代码如下：
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.providers.ssh.operators.ssh import SSHOperator
from airflow.providers.sftp.operators.sftp import SFTPOpera
浏览 3提问于2021-06-25得票数 0
我设置了airflow并运行了一些day，计划每天运行一次"0 0** *“。
我想检查下一次计划运行特定dag的时间，但我看不到在管理员中哪里可以做到这一点。
浏览 1提问于2018-09-07得票数 12
回答已采纳
我有一个运行大约一周的airflow dag-1和一个每天运行几个小时的dag-2。当dag-1正在运行时，由于API限制速率，我不能让dag-2运行(另外，dag-2应该在dag-1完成后运行)。假设dag-1已经在运行，而应该每天运行的dag-2出现故障，我是否可以以正确的方式安排dag依赖关系？是否可以在dag-2应该启动时暂时停止dag-1 (在运行时)，然后在不手动中断的情况下再次运行dag-1？
浏览 16提问于2021-01-25得票数 1
2回答
假设开始日期是2022-03-01。我希望DAG每天早上5点运行。这是我的DAG：
args = {
'owner': 'airflow',
'start_date': datetime(2022, 3, 1),
'depends_on_past': False,
'email': "xxx@xxx.com",
'email_on_failure': False,
'retry': 1,
'retry_
浏览 9提问于2022-03-30得票数 0
回答已采纳
我有一个报告应用程序，它使用芹菜每天处理数千个作业。每种报告类型都有一个python模块，用于封装所有作业步骤。作业采用特定于客户的参数，通常在几分钟内完成。目前，作业是由客户在创建新报告或请求刷新现有报告时按需触发的。
现在，我想添加计划，以便每天运行作业，并自动刷新报告。我知道Airflow在任务编排和调度上大放异彩。我也喜欢将我的工作表达为DAG并从任务重试中获益的想法。我可以看到如何使用Airflow来运行调度的批处理作业，但我不确定我的用例。
如果我将我的作业表示为Airflow DAG，我仍然需要为每个客户运行它们的参数化。这意味着，如果客户创建了一个新报告，我需要有一种方法来触发
浏览 0提问于2020-10-28得票数 0
我在运行Apache气流1.8.1。我希望在我的实例上运行32多个并发任务，但无法使任何配置工作。
我使用的是CeleryExecutor，UI中的气流配置为parallelism和dag_concurrency显示了64，我已经多次重新启动了气流调度程序、web服务器和工作人员(我实际上正在Vagrant中本地测试这个配置，但也在一个EC2实例上进行了测试)。
airflow.cfg
# The amount of parallelism as a setting to the executor. This defines
# the max number of task instances
浏览 0提问于2018-12-05得票数 4
回答已采纳
我正在运行一个DAG，每天运行一次。它从9个并发运行的任务开始，这些任务都做着相同的事情-每个任务基本上都是轮询S3，看看该任务指定的1个文件是否存在。每个任务在Airflow中都是相同的代码，并以相同的方式放入结构中。我有一个这样的任务，在随机的日子里，它不能“开始”-它不会进入running阶段。它就像queued一样。当它这样做的时候，它的日志是这样写的 *** Log file isn't local.
*** Fetching here: http://:8793/log/my.dag.name./my_airflow_task/2020-03-14T07:00:00
**
浏览 200提问于2020-03-25得票数 0
回答已采纳
有没有办法在Airflow DAG中设置/编写自定义schedule_interval？我正在寻找的是一种方法来建立一个时间表，当DAG运行在每天除了节假日(如圣诞节，劳动节，独立日等)。
用标准的cron表达式是不可能实现的。任何帮助/指南都是非常感谢的。
浏览 0提问于2021-01-19得票数 1
我有一个每天运行的DAG，我对它做了一些更改，我需要在几个月内重新运行它。 如果我转到Airflow UI，我可以清除它，它将运行，但这需要大量的点击。有没有其他方法可以做到这一点？如果我从命令行执行回填，如下所示： airflow backfill -s 20200101 -e 20200301 daily_dag_name 这实际上不会运行dag，我相信这是因为它运行成功了，而回填只有在失败时才起作用。
浏览 42提问于2021-01-14得票数 2
回答已采纳
我使用的是Azure VM上的气流。我希望我的DAG每天午夜运行，但我不能总是连接到VM在这个时候运行“气流调度器”命令。我希望我的DAG能够在没有我干预的情况下在午夜运行VM。
我尝试使用Azure VM管理器中的"run命令“，但运行时间限制在90分钟。我也尝试过探索airflow.cfg文件，但是没有找到任何东西。
这是我DAG的心声
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': dat
浏览 0提问于2019-08-21得票数 0
回答已采纳
1回答
假设我有一个计划每天运行的DAG，今天是星期天。有没有办法告诉Airflow重新运行从周一到今天的所有DAG运行？
浏览 12提问于2019-06-21得票数 1
回答已采纳
我有一个简单的气流DAG：
from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.bash import BashOperator
with DAG("Second Dag",
start_date=datetime(2022,1,1),
schedule_interval="0 5 * * *",
catchup=False,
max_active_runs=1
浏览 14提问于2022-02-10得票数 1
我有20个bash脚本。有些每分钟运行一次，每天运行一次，还有一些使用cron每小时运行一次。现在我需要在气流中迁移。为此，根据气流的概念，我需要再创建20个文件(DAG文件)。
airflow是否提供了创建通用dag模板的方法，该模板可以在给定的计划时间内使用不同的dag id执行所有的bash脚本？
我得到了一个参考- ，但我不确定，这是正确的方式还是错误的？
浏览 0提问于2020-01-07得票数 1
2回答
如何正确设置气流调度器
、、、
我试图在airflow中安排任务，但显然不是那么明显。
我想运行的任务非常简单，下面是DAG配置：
from datetime import datetime
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
def print_hello():
return 'Hello world!'
dag = DAG('hello_
浏览 2提问于2018-10-06得票数 1
1回答
我想创建一个dag，它每小时运行一次，我们有一个导入和清理文件的cleanData进程，然后是一个storedata进程，它每天在上午8点和下午6点向我们发送两次报告如何在dag中包含时间？ import pandas as pd
import numpy as np
from airflow import DAG
from datetime import datetime
from airflow.operators.python_operator import PythonOperator
default_args = {
'owner': 'air
浏览 10提问于2021-07-16得票数 1
1回答
我有一个Airflow dag，我需要同时计划它，无论它是标准时间还是夏令时，但airflow服务器和cron计划是基于utc的。它需要在数据库恢复发生后运行，并且该服务器根据本地时间运行恢复。
这导致我不得不将它安排在夏令时的一个小时后。有没有办法让airflow在同一时间运行作业，无论是夏令时还是标准时间？
例如，数据库恢复在每天中午和晚上8点完成。无论是PDT还是PST，这个时间都是一样的。有没有办法在cron中做到这一点，使作业的运行时间始终是中午和晚上8点？
浏览 0提问于2018-09-14得票数 2
我有一个执行Python脚本的DAG，该脚本接受一个日期参数(当前日期)。我计划DAG在周一到周五的早上6:00运行，即工作日的东部标准时间。DAG必须在星期一运行Python脚本，并将星期一的日期作为参数，对于星期二到星期五也是如此，星期五的日期作为参数。 我注意到使用'0 6 * * 1-5'的调度间隔不起作用，因为星期五的执行直到下个星期一才发生。 我将计划时间间隔更改为'0 6 * * *'，使其在每天上午6:00运行，并在dag开始时过滤掉‘0 6 * * 1-5’范围内的日期，因此有效地从星期一到星期五。对于周六和周日，应跳过下游任务。 这是我的代
浏览 60提问于2020-08-18得票数 0
为了获得dag_state，我运行以下LCI命令：
airflow dag_state example_bash_operator '12-12T16:04:46.960661+00:00'
问题是-我必须将确切的日期时间(即execution_date)显式地传递给这个命令。当我运行airflow list_dags时，我只得到DAG的列表，但没有得到它们的执行日期。
有没有办法获得确切的日期时间(i.e. -> '12-12T16:04:46.960661+00:00')？
对于给定的进程，使用命令行CLI？
浏览 0提问于2018-12-12得票数 2
我有一个包含三个bash任务的DAG，计划每天运行。
我想访问所有bash脚本中dag实例的唯一ID (可能是PID)。
有没有办法做到这一点？
我正在寻找与Oozie类似的功能，在那里我们可以在工作流xml或java代码中访问WORKFLOW_ID。
谁能告诉我AirFlow的文档是关于“如何在AirFlow DAG中使用内置变量和自定义变量”？
非常感谢Pari
浏览 1提问于2016-07-11得票数 3
回答已采纳
1回答
上下文:我成功地在sql_alchemy_conn上安装了气流，将executor更改为LocalExecutor，将EC2更改为postgresql+psycopg2://postgres@localhost:5432/airflow；max_threads为10。
我的问题是，当我创建一个每天都要运行的守护进程时，一切都很好，但是当我创建一个dag，它将在周一和周三上午10点运行时，气流不会运行它。有人知道我做错了什么吗?我应该怎么做才能解决这个问题？
，运行良好且正确：
import airflow
from airflow import DAG
from airflow.operato
浏览 4提问于2020-03-24得票数 0
回答已采纳
Airflow老兵请帮帮忙 我正在寻找cron的替代品，偶然发现了apache airflow。 我们有一个设置，其中多个用户应该能够创建他们自己的DAG并计划他们的作业。 我们的用户是可能不知道如何编写DAG python文件的混合用户。此外，他们可能无法访问运行airflow的服务器。 是否可以通过UI创建气流DAG。我找不到任何与此相同的引用。所有示例都讲述了如何创建一个python文件并将其上传到$AIRFLOW_HOME/dag/目录。用户将无权访问此目录。 例如，Rundeck允许用户通过UI添加工作流和任务依赖关系。在airflow中有没有类似的插件/功能。 PS:我真的很喜欢
浏览 155提问于2018-02-26得票数 21
回答已采纳
我创建了一个dag，并每天安排它。它每天都会排队，但任务并不实际运行。这个问题在过去的中已经提出了，但是答案对我没有帮助，所以似乎还有另一个问题。
我的代码在下面分享。我用一个注释替换了任务t2的SQL。当我使用"airflow test...“在CLI上单独运行这些任务时，每个任务都能成功运行。
您能解释一下要让DAG运行应该做些什么吗？谢谢!
这是DAG代码：
from datetime import timedelta, datetime
from airflow import DAG
from airflow.contrib.operators.bigquery_operato
浏览 1提问于2018-07-10得票数 5
回答已采纳
我已经创建了下面的DAG用于每天运行sql脚本。如何将查询结果保存为JSON文件并保存到Google Composer的DAG文件夹中？ import datetime
import airflow
from airflow.operators import bash_operator
from airflow.contrib.operators import bigquery_operator
START_DATE = datetime.datetime(2020, 3, 1)
default_args = {
'owner': 'Alen',
浏览 9提问于2020-03-18得票数 0
我想试着用气流代替Cron。但是schedule_interval并没有像我预期的那样工作。
我编写的python代码如下所示。
以我的理解，气流应该运行在"2016/03/30 8:15:00“上，但它在那个时候行不通。
如果我将其更改为"'schedule_interval'：timedelta(分钟= 5)"，我认为它是正确的。
"notice_slack.sh“只是将松弛api调用到我的通道上。
# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_
浏览 5提问于2016-03-30得票数 18
1回答
我使用的气流1.10.2叉从流行的和使用CLI操作它读取。我已经创建了一个DAG，当我通过UI触发它时，它将被成功地执行。但是，当我使用命令行触发DAG时，DAG状态将开始运行，但任务将永远不会执行。
对于下面图像的引用，DAG的ID是xcom_4_modules_ver_5
在通过调度的进程运行DAG并通过UI手动触发之后，我在CLI上执行airflow list_dag_runs xcom_4_modules_ver_5以检查DAG运行的状态。DAG已成功执行，并将预期的输出写入日志文件中。这对应于ID 1和2。
但是，在执行命令airflow trigger_dag -sd /
浏览 1提问于2019-04-15得票数 8
我有一个Python脚本，它指向elasticsearch集群，对数据执行聚合和计算，然后将洞察存储在本地PostgreSQL中。该脚本可以每天或每周运行，具体取决于用户首选项，如下所示：
python script.py --approach daily
python script.py --approach weekly
我想让这个数据流工作流程通过Airflow每10分钟运行一次。
我的猜测是使用bashoperator创建一个任务t1 = bashoperator和t2 = bashoperator，前者将bash命令python script.py --approach daily
浏览 1提问于2021-02-25得票数 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
回答已采纳
1回答
我正在尝试弄清楚如何配置/调度Airflow DAG，使其在满足条件后每天在准确的时间运行两次，而不是同时运行两次。
我想在午夜和晚上9点运行相同的任务。
为此，我向schedule_interval添加了一个类似0 0,21 * * *的cron，这样它就可以在每天午夜和晚上9点运行但是今天(4月27日)的跑步开始于昨天(4月26日)的00:00:00，同时跑的是00:00:00和21:00:00。
预期行为将在今天(4月27日) 00:00:00运行，21小时后21:00:00再次运行。
有什么想法吗？
最后，问题是:我如何一天运行两次DAG？
谢谢。
浏览 0提问于2017-04-27得票数 0
我已将DAG的执行安排为每天运行。它完美地工作了一天。
然而，每天我不仅想重新执行当天的{{ ds }}，还想重新执行之前的n天(假设n= 7)。
例如，在计划在"2018-01-30“运行的下一次执行中，我希望Airflow不仅使用执行日期"2018-01-30”运行DAG，而且还重新运行从"2018-01-23“到"2018-01-30”的所有前几天的DAG。
有没有一种简单的方法可以使之前的执行“无效”，以便自动运行回填？
浏览 13提问于2018-01-30得票数 5
回答已采纳
我正在尝试运行运行历史时间线之外的过去的特定日期。
例如，附加的 {1}显示了从7月3日到7月8日的每天运行。但是，我们决定需要在6月29日运行一次。
我尝试更改了一些参数，但无济于事，它一直执行到今天7月8日。 {2}
我尝试用以下参数替换：
‘'start_date'：datetime.strptime("2019-06-29"，"%Y-%m-%d")，
schedule_interval="@once“
代码如下：
from datetime import datetime, timedelta, date as dt
浏览 0提问于2019-07-09得票数 0
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
我试图在气流中运行一个简单的dag来执行python文件，它正在抛出错误，无法打开文件‘/User/.’。
下面是我正在使用的脚本。
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime,ti
浏览 9提问于2021-02-25得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
