Airflow -告诉DAG跳过每个月2号的处理
Airflow是一个开源的任务调度和工作流管理平台，它允许用户以有向无环图（DAG）的方式定义、调度和监控任务。在Airflow中，DAG是由一系列任务（Task）和任务之间的依赖关系组成的。
对于需要跳过每个月2号的处理，可以通过在DAG中使用条件语句来实现。具体而言，可以在DAG中定义一个PythonOperator任务，该任务在每次执行时检查当前日期是否为每个月的2号。如果是2号，则直接跳过该任务，否则执行相应的处理逻辑。
以下是一个示例代码：
from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
def process_data():
# 处理数据的逻辑代码
def check_date():
if datetime.now().day == 2:
return 'skip_task'
else:
return 'process_data'
dag = DAG(
'my_dag',
schedule_interval='@monthly',
start_date=datetime(2022, 1, 1)
)
skip_task = PythonOperator(
task_id='skip_task',
python_callable=lambda: None,
dag=dag
)
process_data_task = PythonOperator(
task_id='process_data',
python_callable=process_data,
dag=dag
)
check_date_task = PythonOperator(
task_id='check_date',
python_callable=check_date,
dag=dag
)
check_date_task >> [skip_task, process_data_task]在上述代码中，我们定义了一个名为my_dag的DAG，使用@monthly的调度间隔，从2022年1月1日开始运行。其中，check_date_task任务会根据当前日期决定执行哪个任务，如果是2号，则执行skip_task任务，否则执行process_data_task任务。
需要注意的是，上述代码中并未提及任何腾讯云相关产品，如果需要结合腾讯云的产品进行任务处理，可以根据具体需求选择适合的腾讯云产品，例如使用腾讯云函数（云原生）来执行任务处理逻辑，使用腾讯云数据库来存储数据等。具体的产品选择和介绍可以参考腾讯云官方文档。
相关·内容
我试着在每个月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
1回答
我构建了一个简单的AirFlow管道来处理传入的历史信息。由于数据的性质，每个月的第二天都会有大量的交易量。在那些日子里，我希望以10分钟为增量处理数据：*/10 * 2 * * 剩下的几天，我想正常地处理它，以2小时为增量'0 */2 * * *' 我如何告诉我的“正常”DAG跳过每个月2号的处理？ DAG本身非常简单： dag = DAG(
dag_name,
catchup=True,
default_args=default_args,
schedule_interval=schedule_interval
)
with dag:
historica
浏览 28提问于2019-05-22得票数 0
回答已采纳
1回答
停止气流回填运行
、、
我使用的是Airflow 1.9.0版本。
DAG每分钟都在运行。但是，当气流下降一段时间左右时，我们会看到大量回填作业正在运行，这会停止最近的DAG运行。
我使用的是SubDagOperator。
dag_args = {
'owner': 'Anish',
'depends_on_past': False,
'start_date': datetime.datetime(2017, 3, 20),#time in past
'retries': 0,
'retry
浏览 14提问于2021-06-12得票数 0
1回答
我有一个在MySQL表中插入一些记录的dag，我想每隔2小时运行我的dag。为此，我有以下代码：
from datetime import timedelta, datetime
import pymysql
import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.operators import MySqlOperator
default_args = {
'owner': 'airflow
浏览 54提问于2021-06-04得票数 0
回答已采纳
我有一个任务，我将其称为final，它有多个上游连接。当其中一个上游被ShortCircuitOperator跳过时，这个任务也会被跳过。我不希望跳过final任务，因为它必须报告DAG成功。
为了避免被跳过，我使用了trigger_rule='all_done'，但它仍然被跳过。
如果我使用BranchPythonOperator而不是ShortCircuitOperator，final任务不会被跳过。分支工作流似乎是一种解决方案，尽管不是最优的，但现在final将不会考虑上游任务的失败。
如何使其仅在上游成功或跳过时运行？
ShortCircuit DAG示例：
from
浏览 0提问于2018-08-07得票数 12
2回答
是否可以动态更改DAG的重试次数？
想象一下一个简单的守护进程：
from airflow.models import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
def conditonnal_retry(value=True):
if value:
return "retry should occur if dag run fails"
else:
return "no
浏览 4提问于2020-02-13得票数 0
回答已采纳
airflow.cfg
# airflow version = 1.10.1
executor = LocalExecutor
parallelism = 32
dag_concurrency = 16
和一个dag.py：
with DAG('mydag', schedule_interval="@hourly") as dag:
# define tasks
但有时我的任务比我的schedule_interval需要更长的时间来完成，airflow会在前一项任务完成之前安排下一项任务。
这导致了各种可怕的竞争条件。
有没有一种方法可以明确地防止重叠任
浏览 1提问于2019-07-27得票数 1
我在我的气流DAG中有两个任务。一个触发API调用( Http操作符)，另一个使用另一个api ( Http传感器)不断检查其状态。此DAG计划每小时& 10分钟运行一次。但有时一次执行可能需要很长时间才能完成，例如20小时。在这种情况下，前一个任务正在运行时的所有调度都不会执行。
例如，如果我在01:10完成这项工作，需要10个小时.本应运行的调度02:10、03:10、04:10、... 11:10等被跳过，只执行12:10的调度。
我正在使用本地executor。我正在使用下面的脚本运行airflow服务器和调度程序。start_server.sh
export AIRFLOW_
浏览 11提问于2019-08-09得票数 1
回答已采纳
我有多个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
这是我的气流传感器的样子。
有一个分支任务，它检查一个条件，然后：
直接运行任务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
1回答
我想得到这个DAG的默认args中提到的电子邮件，使用气流中的另一个DAG。我怎么能这么做？请帮帮忙，我是新来的气流！
from airflow.models import DagRun
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
from datetime import datetime, timedelta
from airflow import DAG
def first_function(**context):
pri
浏览 1提问于2022-06-10得票数 0
我试图运行一个库伯内特斯吊舱工作6次。每次它都会打印一个数字并睡眠5秒。然而，它只运行一次，然后就停止了。下面是dag文件的完整代码：
from datetime import datetime, timedelta
from airflow.models import DAG
from airflow.models import Variable
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import (
KubernetesPodOperator,
)
from airflow.operators
浏览 8提问于2022-03-28得票数 0
试图设置一个简单的电报通知，dag立即在没有日志的情况下失败，如果我使用除air字节以外的任何操作员，一切都正常，如果我删除电报操作员仍然工作良好，下面的代码，有人知道原因可能是什么吗？
from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.providers.airbyte.operators.airbyte import AirbyteTriggerSyncOperator
from airflow.providers.telegram.operators.telegram import
浏览 19提问于2022-03-22得票数 0
回答已采纳
我有一个工作流，其中有两个并行进程(sentinel_run和sentinel_skip)，它们应该根据条件运行或跳过，然后连接在一起(resolve)。我需要任何一个sentinel_任务的下游任务都有级联跳过，但是当它到达resolve任务时，resolve应该运行，除非这两个进程在上游出现故障。
基于，"none_failed“触发器规则应该可以工作：
none_failed:所有父母都没有失败(失败或upstream_failed)，即所有父母都成功或被跳过
这也是对的回答。
然而，当我实现一个微不足道的例子时，我看到的不是这样的：
from airflow.model
浏览 9提问于2019-10-09得票数 3
回答已采纳
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成功，我希望执行任务2，如果任务1失败，我希望运行任务3，如果需要，我希望分配另一个流程。
基本上，我想在没有ssh操作符的气流中运行有条件的任务。
from airflow import DAG
from airflow.operators import PythonOperator,BranchPythonOperator
from airflow.operators import BashOperator
from datetime import datetime, timedelta
from airflow.models import Variable
def t2_e
浏览 0提问于2018-04-18得票数 4
回答已采纳
我得到了以下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
回答已采纳
我制作了一个非常简单的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
回答已采纳
3回答
我得到了关于Airflow的一般查询
是否可以基于另一个dag文件的调度来调度一个dag文件。
例如，如果我有两个dag2，即dag1和dag2。我正在尝试是否可以在每次dag1成功时都运行dag2，否则dag2就不会运行。在气流中这是可能的吗。
浏览 0提问于2018-05-31得票数 5
1回答
我有一个文件定义了DAG对象：
dags/my_dag.py
from airflow import DAG
from datetime import datetime
default_args = {
'owner': 'pilota',
'depends_on_past': False,
'start_date': datetime(2019, 10, 1),
'email': ['some@email.com'],
'email_on_fail
浏览 4提问于2019-10-01得票数 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
1回答
我在本地安装了airflow，因为我正在测试airflow (2.0.0)中的sftp操作符。当我尝试运行这段代码时 from airflow.providers.sftp.operators import sftp_operator
from airflow import DAG
import datetime
dag = DAG(
'test_dag',
start_date = datetime.datetime(2020,1,8,0,0,0),
schedule_interval = '@daily'
)
get_operation = SFTPO
浏览 68提问于2020-12-29得票数 0
回答已采纳
有没有办法在Airflow DAG中设置/编写自定义schedule_interval？我正在寻找的是一种方法来建立一个时间表，当DAG运行在每天除了节假日(如圣诞节，劳动节，独立日等)。
用标准的cron表达式是不可能实现的。任何帮助/指南都是非常感谢的。
浏览 0提问于2021-01-19得票数 1
要求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
在airflow.cfg文件中，我将min_file_process_interval值配置为120秒。达格在每分钟后都有时间间隔运行。但是，它只是在每120秒之后(根据min_file_process_interval值)才被调度。这是意料之中吗？
我将min_file_process_interval更改为200秒，然后在200秒后开始执行dag计划。澄清一下，如果情况相反，即dag调度间隔为2分钟，min_file_process_interval为1分钟，则dag将按照其日程运行良好。下面是我的爸爸：
from airflow import DAG
from airflow.opera
浏览 0提问于2019-05-14得票数 0
回答已采纳
2回答
我有一个DAG，它有5个并行任务。我不想一蹴而就。我想将一些标志/值传递给任务，假设如果将标志设置为True，它将运行，如果标志设置为False，则会跳过。我们能用Airflow2做这个吗？默认参数是非常基本的参数，我没有安排DAG。
我的DAG流看起来像这样
starttask>>5paralleltask>>endtask
我用for循环创建了这5个并行任务
提前谢谢你
浏览 3提问于2022-03-30得票数 0
回答已采纳
我们有气流运行(使用码头组合)和几个DAG的活动。上周，我们将气流更新为2.1.3版。这导致使用DockerOperator的DAG出现错误：
airflow.exceptions.AirflowException: Invalid arguments were passed to DockerOperator (task_id: t_docker). Invalid arguments were:
**kwargs: {'volumes':
我发现这个告诉我
将airflow.providers.docker.operators.docker.DockerOperator和
浏览 5提问于2021-09-10得票数 1
回答已采纳
1回答
我正在尝试设置一个DAG，其中一个任务每分钟运行一次，然后另一个任务在第5分钟(就在1分钟任务之前)运行。这实际上只是测试，我不打算在这么短的时间内运行作业。
在视觉上，我的DAG看起来是这样的：
代码本身是这样的：
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import BranchPythonOperator
from datetime import datetime, timedel
浏览 0提问于2018-10-18得票数 1
回答已采纳
我面临一些问题，试图建立一个基本的DAG文件在气流(但我也有另外两个文件)。
我通过Ubuntu使用LocalExecutor，并将我的文件保存在“C：\Users\tdamasce\Documents\工作区”中，其中包含了dag和日志文件。
我的剧本是
# step 1 - libraries
from email.policy import default
from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.dummy_operator import DummyO
浏览 5提问于2022-03-24得票数 0
回答已采纳
2回答
我刚开始使用。我试图运行一个进程，不想做任何调度。
我希望使用命令行参数运行管道，并覆盖所有当前输出。我没有开始日期，没有调度，没有时间，也没有重试逻辑，我只是想按顺序运行一组函数来开始。
文档总是包含一个日期。
airflow test tutorial print_date 2015-06-01
我希望运行这个守护进程，以便它执行所有函数，并忽略以前的任何运行。如何从我的进程中删除所有日期和日期逻辑？
我有一个修改过的教程守护文件版本：
"""
Code that goes along with the Airflow tutorial located at:
ht
浏览 6提问于2016-12-10得票数 2
回答已采纳
AIRFLOW_HOME=/path/to/my/airflow_home 我得到这样的警告... >airflow trigger_dag python_dag3
/Users/alexryan/miniconda3/envs/airflow/lib/python3.7/site-packages/airflow/configuration.py:627: DeprecationWarning: You have two airflow.cfg files: /Users/alexryan/airflow/airflow.cfg and /path/to/my/airflow_hom
浏览 44提问于2019-12-24得票数 3
我有一个达格，每小时安排一次。假设凌晨1:00，凌晨2:00，凌晨3:00假设02:00被选中，但是如果01:00am dag运行仍在进行中，则需要取消02:00am实例。
我在尝试这个密码。
local_tz = pendulum.timezone("America/Chicago")
default_args = {
'owner': 'airflow',
'depends_on_past': True,
'start_date': datetime(2020, 11, 15, tzi
浏览 15提问于2020-11-17得票数 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
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
回答已采纳
我有一个文件路径为
fpath = "/path/abc20200216"
abc之后的日期每个月都会改变。
现在我想在FileSensor中使用它
file_sensor = FileSensor(task_id= "file_sensor_job", poke_interval= 30, filepath = fpath, run_as_user='xyzzy',dag=dag)
如何在运算符中传递通配符关键字。
或者，是否有其他方法可以使用airflow感测远程服务器中的文件？
浏览 15提问于2020-02-26得票数 2
我试图在subdag创建时从父进程访问一些xcom数据，我在互联网上搜索以实现这一点，但是我没有找到任何东西。
def test(task_id):
logging.info(f' execution of task {task_id}')
def load_subdag(parent_dag_id, child_dag_id, args):
dag_subdag = DAG(
dag_id='{0}.{1}'.format(parent_dag_id, child_dag_id),
default_args=
浏览 0提问于2019-02-18得票数 8
回答已采纳
我目前正在试验一种新的概念，即操作员将与外部服务通信以运行操作员，而不是在本地运行操作员，外部服务可以与气流通信以更新DAG的进度。
例如，假设我们有一个bash操作符：
bash_task = BashOperator(
task_id="bash_task",
bash_command="echo \"This Message Shouldn't Run Locally on Airflow\"",
)
这是DAG的一部分：
from airflow import DAG
from airflow.operators.
浏览 2提问于2022-08-10得票数 0
回答已采纳
在Apache Airflow中，是否可以捕获失败的bash命令生成的原始错误消息，而不是Apache Airflow生成的跟踪错误，它会告诉您线路失败，但不能确切地说明失败的原因？
Dag中的示例行：
gsutil_rsync = BashOperator(
task_id="task1",
bash_command='gsutil rsync -r s3://bucket/ gs://bucket',
dag=dag)
浏览 2提问于2018-10-12得票数 2
3回答
关于“动态任务”的其他问题似乎解决了在计划或设计时动态构建DAG的问题。我对在执行过程中将任务动态添加到DAG很感兴趣。
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
dag = DAG('test_dag', description='a test',
浏览 2提问于2018-02-03得票数 21
回答已采纳
1回答
“开始日期”：日期时间(2020年，1，1)，
时区=“亚洲/新加坡”，
如果我想每周三上午00:55运行一个报告，我应该输入‘550** 3'，但是我必须输入’550** 4‘。(这样，执行时间将如预期的那样，每周三00:55中士)
类似地，如果我希望每个月第一天运行第3天AM SGT，我应该输入‘schedule_interval =’031* *'，但是它会将它排在第3个月的最后一天。
怎么一回事？
我知道UTC的时间是提前8小时，但这仍然不能解释这一点，因为这些开始前一天，他们的计划开始时间在中科院。因此，对于下面的内容(仅在相关部分中复制)，DAG运行时都是周三上午0
浏览 3提问于2020-07-11得票数 0
回答已采纳
我有一个执行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来调用一个emr实例，使用submit。在设置spark_test_steps时，我需要包括从POST Json中传入的变量，以填充火花提交，如下所示：
SPARK_TEST_STEPS = [
{
'Name': 'calculate_pi',
'ActionOnFailure': 'CONTINUE',
'HadoopJarStep': {
'Jar': 'com
浏览 3提问于2019-09-19得票数 0
回答已采纳
我在蔚蓝的库伯奈特斯部署了阿帕奇气流。
apache气流的Helm存储库：
AKS版本: 1.16.13
一旦我部署了气流，我就用这个程序测试了它：
from airflow import DAG
from datetime import datetime, timedelta
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
from airflow.operators.dummy_operator import DummyOperator
default_args =
浏览 0提问于2020-10-08得票数 1
1回答
我正在研究Airflow文档，以便更好地了解它的调度机制。我在下面遇到了一个例子。 文件中提到，当调度人员在2016-01-02上午6点选择DAG时，将创建一次DAG运行，execution_date为2016-01-01，下一次运行将在2016-01-03凌晨创建，执行日期为2016-01-02。 调度间隔是按小时提供的，执行日期是指DAG在结束时运行的时间段的开始，那么为什么不是在2016-01-02早上6点调度器挑选DAG的前一个小时呢？ """
Code that goes along with the Airflow tutorial located at:
浏览 77提问于2021-04-25得票数 1
1回答
在处理多个环境(如Dev/Staging/Prod等)时，为每个环境设置单独(最好是相同的)气流实例将是最好的情况。
我正在使用GCP管理的气流()，这不是很便宜的运行，有多个实例将大大增加我们的每月账单。
那么，我想知道是否有人建议使用单个气流实例来处理多个环境？
我正在考虑的一种方法是在我的dags folder中有对应于每个环境的单独的顶级文件夹(即DAG /dev、dags/prod等)，并通过CI/CD管道将我的DAG脚本复制到相关文件夹中。
因此，在我的源代码存储库中，如果我的守护进程看起来像：
airflow_dags/my_dag_A.py
在CI阶段，我可以有一个构建步骤来创
浏览 6提问于2022-12-03得票数 0
我正在尝试了解Airflow是否支持在DAG中跳过subDags。
假设我的DAG是这样的: task1(FileSensor) -> task2(subDag-1) -> task3(subDag-2) -> task4(subDag-3)
假设我有一个想要运行的subDags的列表，比如['1','3']，这意味着我想运行task1，然后运行task2，然后运行task4，跳过task3。(触发DAG时，将通过配置提供待运行的subDags列表)
我已经研究过使用PythonOperator跳过任务，但我不想在PythonOperator中
浏览 0提问于2021-03-04得票数 0
1回答
我对气流很陌生，但我设法在本地运行它(用SQLite)跟踪文档。
考虑到我需要一些额外的工具，我试着把所有的东西都放入一个码头，但我没有做到。
我删除了添加的所有内容，但仍然找不到问题。然后，我遵循了这个，将文件的版本从3.8降到3.3，并删除了重新启动策略；但我甚至达不到端口8080。
更进一步，我找到了这个，它允许我到达端口8080，但是当我尝试使用命令行创建一个用户时，它会从数据库中返回一个错误，告诉我不存在名为log或dags的表。通过UI创建用户可以正常工作，但我觉得奇怪的是，db没有看到这一点(我检查了db容器日志)。
一个小的例子是下面的docker-compose.yml
ve
浏览 14提问于2021-02-08得票数 1
我创建了DAG，如下所示：
args = {
'owner': 'airflow',
'start_date': datetime(2018, 12, 30),
'retries': 5,
'retry_delay': timedelta(minutes=30)
}
dag = DAG(
dag_id='TEST',
default_args=args,
schedule_interval=timedelta(minutes=30),
浏览 1提问于2018-12-24得票数 1
回答已采纳
我使用Python调用天气API，但是气流任务由于错误Negsignal.SIGSEGV而失败。调用天气API的Python在气流外部运行时运行良好。
DAG
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperato
浏览 40提问于2022-09-02得票数 3
回答已采纳
按照教程，我创建了一个文件夹$AIRFLOW_HOME/ DAG，并将教程DAG文件放在那里。然后启动气流调度器。默认情况下，它是暂停的。但是如果我看一下气流调度器的输出，我看到了很多运行，试图创建DAG。它为什么一直跑？
[2018-09-10 15:49:24,123] {jobs.py:1108} INFO - No tasks to consider for execution.
[2018-09-10 15:49:24,125] {jobs.py:1538} INFO -
========================================================
浏览 1提问于2018-09-10得票数 8
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
