气流示例DAG在图形用户界面中需要很长时间，调度程序显示如下: DagFileProcessorManager (PID=...)上次发送的心跳信号
气流示例DAG是指在Apache Airflow中使用的有向无环图（Directed Acyclic Graph，DAG）的一个示例。在图形用户界面（GUI）中，如果气流示例DAG需要很长时间来加载，可能是由于以下原因导致的：
- DAG文件处理器管理器（DagFileProcessorManager）的进程ID（PID）在显示的心跳信号之前已经结束或崩溃。这可能是由于进程异常终止、资源不足或其他问题导致的。在这种情况下，可以尝试重新启动DagFileProcessorManager进程或检查相关日志以了解更多详细信息。
- DAG文件处理器管理器正在处理大量的DAG文件或复杂的DAG定义，导致加载时间较长。这可能是由于DAG文件数量过多、DAG定义中包含大量任务或依赖关系复杂等原因导致的。在这种情况下，可以考虑优化DAG定义，减少任务数量或简化依赖关系，以提高加载性能。
- DAG文件处理器管理器所在的服务器或网络存在性能瓶颈，导致加载时间延长。这可能是由于服务器资源不足、网络带宽限制或其他网络问题导致的。在这种情况下，可以尝试增加服务器资源、优化网络配置或与网络管理员协商解决问题。
对于气流示例DAG在图形用户界面中加载时间过长的问题，可以考虑以下解决方案：
- 检查DagFileProcessorManager进程是否正常运行，并重新启动该进程。
- 优化DAG定义，减少任务数量或简化依赖关系。
- 增加服务器资源，提高服务器性能。
- 优化网络配置，确保网络带宽充足。
- 与Airflow社区或相关技术支持团队联系，寻求更详细的帮助和指导。
腾讯云提供了一系列与云计算相关的产品，包括但不限于：
- 云服务器（CVM）：提供灵活可扩展的云服务器实例，用于部署和运行应用程序。 产品介绍链接：https://cloud.tencent.com/product/cvm
- 云数据库MySQL版（CDB）：提供高性能、可扩展的云数据库服务，支持MySQL数据库引擎。 产品介绍链接：https://cloud.tencent.com/product/cdb_mysql
- 云原生容器服务（TKE）：提供基于Kubernetes的容器化应用管理平台，简化容器部署和管理。 产品介绍链接：https://cloud.tencent.com/product/tke
- 人工智能机器学习平台（AI Lab）：提供丰富的人工智能开发工具和资源，支持机器学习和深度学习任务。 产品介绍链接：https://cloud.tencent.com/product/ailab
请注意，以上仅为腾讯云的部分产品示例，更多产品和详细信息可以在腾讯云官方网站上找到。
相关·内容
我正在学习Apache气流，并尝试编写hello world版本的它。
我已经用python版本3.8和下面的包含dag和任务的代码安装了airflow版本2.0
import datetime
import logging
from airflow import DAG
from airflow.operators.python import PythonOperator
def greet_hello():
logging.info("Hello World")
dag = DAG("FirstDag", start_date=dateti
浏览 7提问于2021-02-07得票数 2
回答已采纳
在我的例子中，我在dags路径下编写了一个dag文件。启动气流调度程序后，它将成功地加载dag文件。但是，在更改了dag文件后，它失败了加载数据文件。是否有任何建议，不重新启动调度程序loade dag文件？
浏览 5提问于2017-09-26得票数 2
回答已采纳
按照教程，我创建了一个文件夹$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
回答已采纳
2回答
任务间气流延迟
、、、
正如您在图像中看到的那样：
气流在任务执行之间花费了太多的时间？它几乎占DAG执行时间的30%。我已将airflow.cfg文件更改为：
job_heartbeat_sec = 1
scheduler_heartbeat_sec = 1
但我的潜伏期还是一样的。
它为什么会这样表现？
浏览 0提问于2018-04-18得票数 16
回答已采纳
1回答
我尝试记录在终止postgres挂起查询的气流中触发我的DAG的用户，但是它不起作用。你能帮帮忙吗？我遗漏了什么？当我在气流中检查日志而不是用户名时，到处都是‘无’。
utils.py (描述会话逻辑的地方)
import logging
from airflow.models.log import Log
from airflow.utils.db import create_session
from airflow.operators.python_operator import PythonOperator
from psycopg2.extras import RealDictCurs
浏览 6提问于2021-08-11得票数 0
回答已采纳
1回答
我对气流很陌生，真的很感激对以下问题的任何帮助。我试着在笔记本电脑上运行气流网络服务器。
在理论上，我设置了start_time=datetime.now()，当我在set服务器上手动运行这个进程时，应该成功地运行它，但是它改变了超时，它要么排队，要么成功。有时它是成功的(但运行时是00:00:00，显然我的进程没有运行)，有时只是排队而已。
这是我的DAG中的代码：
from datetime import datetime
from airflow import DAG
from airflow.models import Variable
from airflow.operators.p
浏览 8提问于2022-10-31得票数 0
1回答
我的守护进程大约需要50秒来解析，我只使用外部触发器来启动dag运行，没有调度。我注意到，气流想要在每个trigger_dag命令中填充很多->，在后台，它一直在检查dags文件夹，并在新的.py部署后立即创建.py文件。
不管怎么说，我可以部署我的集群，并得到填补一次！然后在接下来的2周里，在任何trigger_dag上立即开始运行(现在只需要50秒才能在开始之前填满达格袋)。我不需要在2周内更新dag定义。
浏览 0提问于2019-04-25得票数 7
回答已采纳
在同一个DAG中，任务之间的等待时间很长，我遇到了问题。我在运行气流1.10。有人能指出哪些设置与调整相关吗？请参阅下面的例子。
这里最长的等待时间超过一个半小时。
浏览 2提问于2018-09-26得票数 3
回答已采纳
2回答
各位，我正在使用气流(1.10.12节)的一个项目，安排乔布斯的日常工作。希望并行执行多个DAG(14)。更新了cfg文件中的并发参数。但是，已经观察到了这样的行为:调度程序开始执行任务，将任务分配给队列，但经过一定的时间暂停，5分钟后重新启动。由于这种行为，所有Dag所花费的时间都很长，而且有些任务也会失败。有人能帮我理解为什么调度器停下来，如果我们应该修改一些参数或转移到其他气流版本？
下面的是配置文件
[core]
dags_folder = /home/airflow/dags
base_log_folder = /home/airflow/logs
remote_logging =
浏览 8提问于2020-11-19得票数 0
我们在GCP ( Kubernetes集群上的管理气流)中使用Composer来调度ETL管道。
我们的DAG (200-300)是动态的，这意味着所有的DAG都是由单个生成器DAG生成的。在气流1.x中，由于调度器的限制，它是一种反模式。但是，调度器在气流2.x中处理此场景更好。参见3点。
我们有一个非常强大的环境(请参阅下面的技术细节)，但是我们在任务更改之间遇到了很大的延迟，这对调度程序来说是个坏兆头。此外，队列中还有许多任务在等待，这对工人来说是个坏兆头。当50-60个DAG被触发并运行时，会出现这些性能问题。在我看来，这种并发性并没有那么大。
我们使用的是Composer，它根据具有
浏览 3提问于2022-04-05得票数 0
请不要否决我的答案。如果需要，我将更新和纠正我的话。我做了我的家庭工作研究。我并不是新来的，所以试着去理解这一点。
我想了解Google云平台上的气流是如何从dags主文件夹到UI的变化的。此外，请帮助我的达格安装脚本。我和书一起读了这么多答案。图书链接是
我试着从第69页找出我的答案。
3.11调度&触发气流调度器监视所有任务和所有DAG，并触发满足依赖关系的任务实例。在幕后，它监视可能包含的所有DAG对象并与文件夹保持同步，并定期(每分钟左右)检查活动任务，以确定它们是否可以被触发。
我从这本书中了解到，调度程序经常从dags主文件夹中获取更改。(是对的吗？)
--我还在堆
浏览 0提问于2020-08-23得票数 3
回答已采纳
我们使用的是气流版本2，当我们将DAG文件放置在DAG文件夹中时，需要20到30秒才能在气流数据列表中反映这个dag。
注意事项：我们使用芹菜执行器和postgres作为数据库。
有人能帮我如何使气流更快地拿起DAG吗？它是否有一个配置/任何想法都是值得注意的。
谢谢你，哈利
浏览 3提问于2021-09-19得票数 1
回答已采纳
我对气流很陌生，我正在用GoogleCloudStorageDownloadOperator尝试一些简单的东西：
default_args = {
'start_date': airflow.utils.dates.days_ago(0),
'schedule_interval': None,
'retries': 1,
'retry_delay': timedelta(minutes=5),
'params': {
'work_dir': &
浏览 0提问于2020-01-09得票数 1
回答已采纳
1回答
我有以下的文件夹结构在气流达格文件夹。
/opt/airflow/dags
--my_module/
--my_dag.py
Dag文件(my_dag.py)需要my_module中的python脚本来执行工作流任务。
上面的设置运行得很完美，但是当更新"my_module“文件夹中的python脚本时，它不会反映在工作流中。
当更改dag文件时，它将反映气流中的变化。我也重新启动了气流，但没有运气。我在这里做错了什么..。
浏览 0提问于2018-08-28得票数 0
回答已采纳
假设我有一个气流数据文件，它创建了一个这样的图形.
def get_current_info(filename)
current_info = {}
<fill in info in current_info relevant for today's date for given file>
return current_info
files = [
get_current_info("file_001"),
get_current_info("file_002"),
....
]
for
浏览 6提问于2019-10-26得票数 1
回答已采纳
我正在上使用气流(版本: composer-1.10.2- airflow -1.10.6)。
我意识到，当有许多任务要处理时，调度程序不会调度任务(参见下面的甘特视图)
(不要注意颜色，红色的任务是"createTable操作符“，如果表已经存在，就会失败，所以在下一部分(重要的部分)运行之前，它们必须失败5次)
任务之间有时间间隔！(例如，上午10时至下午15时5小时，什么也没有发生)
正常情况下，它可以很好地工作~40个DAG，每个任务约100到200个任务(有时更多)。但是最近我增加了两个DAG，它有很多任务(每个任务大约5000)，调度程序非常慢，或者没有调度任务。在屏
浏览 3提问于2020-06-04得票数 3
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
我刚做了个气流DAG。当我看到airflow最后一次运行日期时，它会显示最后一次运行日期。当我将鼠标悬停在"i“图标上时，它会引起我的注意，因为它显示了正确的日期。有什么办法可以解决这个问题吗？听起来像是胡说八道，但我最终还是用它来对我的数据进行QA。
浏览 29提问于2019-07-25得票数 2
回答已采纳
6回答
我有一个dag，它检查要生成的新工作流(动态DAG)，如果找到，就创建它们。(参考文献： )
上面的DAG正在工作，动态DAG正在被创建并列在web服务器中。这里有两个问题：
在web url中单击DAG时，它会说"DAG似乎丢失了“。
列出的DAG没有使用“气流list_dags”命令列出。
错误：
达格"app01_user“似乎不见了。
其他所有动态生成的DAG也是如此。我编译了Python脚本，没有发现任何错误。
Edit1:我试着清除所有数据并运行“气流运行”。它成功运行，但没有动态生成的DAG添加到“气流list_dags”。但是当运行命令&
浏览 7提问于2019-08-13得票数 10
1回答
我使用的是气流和Composer，由于我对气流调度器(它是慢的或停止的)有一些问题，版本:composer-1.10.4-气流-1.10.6
我推出了一个“巨大”的收集(因为我有时会需要它)与气流测试我的管道的可伸缩性。其结果是，我的调度程序显然只调度很少任务的DAG，而大型DAG的任务没有被调度。你对此有什么见解或建议吗？
下面是有关我当前配置的信息：
集群配置：
10个集群节点，20个vCPU，160 10内存
气流配置：
核
store_serialized_dags:真
dag_concurrency: 160
store_dag_code:真
min_fi
浏览 5提问于2020-06-11得票数 1
我有两个DAG，需要用Airflow 1.10.2 + CeleryExecutor来运行。第一个DAG (DAG1)是从s3到Redshift的长时间数据加载(3+小时)。我的第二个DAG (DAG2)对DAG1加载的数据执行计算。我希望在DAG2中包含一个ExternalTaskSensor，以便在数据加载后可靠地执行计算。理论上这么简单！ 通过确保两个DAG1被安排为同时启动(两个DAG1的schedule=为“0 8 *”)，并且DAG2依赖于DAG1中的最后一个任务，我可以成功地让DAG2等待DAG1完成。但是当我引入传感器的时候，我发现DAG1上的ETL有很大的延迟。起初，我认为
浏览 366提问于2019-05-21得票数 2
我非常困惑的是，气流如何在新的DAG代码与调度器的变化。有人能澄清气流调度器如何与新代码一起工作吗？每次更改DAG中的代码时，是否都需要停止并重新运行airflow scheduler？或者，我可以设置--num_runs 1并在每次进行新更改时运行它吗？
谢谢!
浏览 0提问于2018-07-17得票数 2
回答已采纳
我正在开发一个基于容器的Apache气流应用程序。我的环境由以下组件组成：
气流调度器
气流网络服务器容器
气流芹菜花容器
气流工人容器(1)
等。
我对这个模式的理解是，我可以拥有一个调度器和一个webserver容器，这些容器只需要一个气流所必需的依赖项，然后我就可以有一个工作节点(或多个)来运行我的DAG所需的一切。
当我尝试以这种方式使用它时(例如，在worker节点中添加和使用一个模块，假设它是crypto模块)，我在前端得到一个DAG Import Error异常，它是这样的：ModuleNotFoundError: No module named &#
浏览 1提问于2021-04-29得票数 3
1回答
我已经在气流中为python中的函数编写了dag脚本，在dag脚本中，我导入了python脚本，作为"from data_extractor.py import *“，.I为data_extractor_dag.py.But中的data_extractor.py中的函数构建了一个工作流，当我在data_extractor.py中进行任何代码更改时，风流守护程序data_extractor_dag.py无法在这个场景中使用load.During，在这个场景中，气流UI页面也不是loaded.How以更好的方式使用气流。
浏览 0提问于2018-10-11得票数 1
我使用的是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
回答已采纳
我是airflow的新手，我尝试通过UI手动触发作业。当我这样做的时候，调度程序继续记录它是Failing jobs without heartbeat的，如下所示：
[2018-05-28 12:13:48,248] {jobs.py:1662} INFO - Heartbeating the executor
[2018-05-28 12:13:48,250] {jobs.py:1672} INFO - Heartbeating the scheduler
[2018-05-28 12:13:48,259] {jobs.py:368} INFO - Started process (PID
浏览 4提问于2018-05-28得票数 6
1回答
实际上，在触发JSON之前和在DAG脚本中，我在气流中传递了JSON{“列”："ABC123"}。
我用DAG脚本编写了如下代码
variable1 = "['Column']"
variable2 = variable1[3:6]
print( {variable2} )
实际上，我想在气流日志中将值打印为123，但是它在logs...DAG中没有打印成功，但是不能在日志中打印值，无论我在触发DAG时传递给哪个列。
浏览 2提问于2021-07-21得票数 0
回答已采纳
我有一个关于Apache气流和Metaflow()之间的区别的问题。据我所知，Apache气流只是一个作业调度程序，它运行任务。Netflix的Metaflow是一个数据流库，它以DAG的形式创建机器学习管道(可用数据流)。基本上意味着，Metaflow可以在Apache气流上执行吗？
我的理解正确吗？如果是，是否可以将Metaflow DAG转换为Apache气流DAG？
浏览 7提问于2022-01-03得票数 2
回答已采纳
1回答
我有一个作为web服务运行的应用程序，它在用户请求时将作业提交给Spark。每个用户都需要限制作业队列。我计划使用气流作为业务流程框架来管理作业队列，但是虽然它支持并行DAG执行，但它是针对批处理而不是实时优化的。气流是设计用来处理每秒200次DAG执行的多个队列(每个用户一次)，还是我应该寻找替代方案？
浏览 0提问于2018-02-10得票数 6
我暂停了一个月的工作。这项工作每10分钟进行一次。现在，我打开了它，我可以看到，从我停止工作的那天起，气流正在尝试运行所有的东西。
每次我要清除任务列表时，就会有更多的任务被排定。它可能有2000+任务。
我希望DAG启动当前任务，并丢弃过去的所有任务。
浏览 1提问于2020-11-20得票数 0
回答已采纳
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
回答已采纳
1回答
步骤(今日为2020年1月7日)
( 1)在气流中放置以下参数：
from datetime import datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
dag = DAG(dag_id='example_dag', start_date=datetime(2020, 1, 1), catchup=False)
t1 = BashOperator(task_id='bash_task', bash_command='e
浏览 3提问于2020-01-07得票数 0
回答已采纳
我有一组按@monthly计划从2011-01-01运行到2018-04-01的DAG。为此，我在每个dags上设置了start_date和end_date。然而，我现在意识到我需要它再运行一个月。我已经将end_date更新为2018-05-01，但是更新这个值并没有触发Airflow启动另一个DagRun。
我已经阅读了，但我不想更改Dag id，因为这将导致我不得不重新运行所有月份，我只想添加一个额外的月。在气流中这是可能的吗？
浏览 0提问于2018-05-26得票数 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
回答已采纳
1回答
我是新来的气流，并试图运行一个简单的DAG每小时的费率。schedule_interval='@hourly'不起作用。我的期望是它将触发DAG从我的start_date (2022年10月7日凌晨)到现在每小时，因为追赶是默认设置为True。但是，它只触发一次z DAG，然后停止。另外，在用户界面上，日程表上写的是1 day，而不是hour。下面给出了代码，UI的截图也附在了后面。气流版本为2.4.1
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime i
浏览 3提问于2022-10-08得票数 1
在我们的大数据项目中，有3000个表需要加载，所有这些表都应该由气流中的一个单独的DAG来处理。
在我们的解决方案中，单个python文件生成每种类型的表加载器，因此它们可以通过基于事件的REST通过云函数单独触发。因此，我们使用以下方法生成我们的DAG：
用于DAG生成器逻辑scd2的气流变量生成表类型的表名列表:插入追加、截断加载、scd1、
表加载器DAG的特定运算符使用的RawToRawRC_TableN 气流变量，例如:用于操作器处理的RR_TableN = {} // python = {} / python用于操作处理RawToCuration
用户定义宏:在任务定义之间，我们尝
浏览 3提问于2021-11-06得票数 1
回答已采纳
1回答
目前，我使用的气流版本: 1.10.10
打开到airflow/logs文件夹后，有许多文件夹是根据DAG名称命名的，但是有一个名为调度程序的文件夹，当打开该文件夹时，包含以日期格式命名的文件夹(例如，2020/07/08 )，直到我首次使用airflow.After搜索多个论坛时，我仍然不确定该文件夹日志用于干什么。
无论如何，问题是我一直在想，删除调度器文件夹中的内容是否可以，因为它占用了很大的空间，不像根据DAG名称命名的文件夹的其余部分(我假设每个DAG运行的日志都存储在那里)。删除调度程序的内容是否会导致DAG日志的错误或丢失？
这可能是一个愚蠢的问题，但我想确定，因为气流在生产服务
浏览 3提问于2020-07-08得票数 5
回答已采纳
我目前正在试验一种新的概念，即操作员将与外部服务通信以运行操作员，而不是在本地运行操作员，外部服务可以与气流通信以更新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
回答已采纳
我们最近从风流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
我在一个团队中工作，使用一个大型云提供商来管理我们所做的事情。每天早上在我开始工作之前，我都有一份有计划的工作，它在云中支撑着一个开发环境，每天晚上，我都有一项计划好的工作，它再次把一切都搞砸了。该开发环境包括Apache气流的一个实例，而作业所做的另一件事是运行一个包含一个任务的气流DAG。对于这个DAG，我有一个间歇性的问题，DAG会运行，但是偶尔该任务的任务实例无法被调度。今天上午发生了这种情况，下面是任务实例的详细信息：
在这种情况下：
调度程序正在运行，并且绝对不是在重负载下运行(没有其他的运行)
据我所知，它还没有运行
我有一个简单的方法来解决这个问题，我会重新启
浏览 4提问于2020-04-17得票数 2
1回答
我对气流很陌生。
我安排一个dag文件每10分钟运行一次。我的默认args在下面。
args = {
'owner': 'crv',
'start_date': datetime(2021, 1, 1),
}
dag = DAG(
dag_id='score',
default_args=args,
schedule_interval='*/10 * * * *',
tags=['example']
)
我使用以下命令在后台运行气流调度程序。
air
浏览 0提问于2021-01-29得票数 1
回答已采纳
我正在使用Apache气流版本1.10.3与顺序执行器，我希望DAG在一定时间后失败，如果它还没有完成。我尝试在示例代码中设置dagrun_timeout
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
default_args = {
'owner': 'me',
'depends_on_past': False,
'
浏览 4提问于2020-02-21得票数 7
回答已采纳
2回答
不幸的是，即使在阅读了这里的许多问题和FAQ页面的气流网站，我仍然不明白气流如何安排任务。我在这里有一个非常简单的示例任务：
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
default_args = {
"depends_on_past": False,
"start_date": datetime(2020, 5, 29),
"e
浏览 4提问于2020-05-29得票数 1
回答已采纳
我使用Kubernetes安装了气流，并登录了气流用户界面。它显示了所有的达格，但它们没有正确显示。
1/在Dag名称的左边没有开/关按钮，它只是显示空复选框。
2/“最近的任务”和"DAG运行“列看起来像是在加载什么；
3/如果我单击并因此进入任何DAG，它似乎试图加载什么；
我尝试了两个气流2.0.0和1.10.11，他们显示相同，所以这不是因为版本。
气流的问题是什么，如何解决？
-在这里，我根据Ofek Hod的建议提供更多的信息: 1/运行"kubectl日志<pod_id> webserver"，在我登录airflow web
浏览 2提问于2020-10-16得票数 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，该dag在Airflow UI中可用，并将其打开以运行它。运行dag后，状态显示为可重试。在那之后，我去了服务器，使用了命令"Airflow scheduler“，然后dag就成功了。
在运行dag之前，调度程序已经启动并运行，我不确定为什么会发生这种情况。当我们创建一个新的dag时，我们需要运行气流调度器吗？想知道调度器是如何工作的。
谢谢
浏览 0提问于2020-05-20得票数 1
在气流中，我应该如何处理错误“此DAG在webserver DagBag对象中不可用，它显示在此列表中，因为调度程序将其标记为元数据数据库中的活动”？
我将一个新的DAG复制到一个气流服务器上，并尝试：
取消暂停并刷新它(基本操作过程，在前面的答案中给出)
重新启动the服务器
重新启动调度程序
停止webserver和调度器，重置数据库(airflow resetdb)，然后再次启动webserver和调度程序。
运行airflow backfill (这里建议使用)
运行airflow trigger_dag
调度程序日志显示它正在处理，没有出现错误，我可以
浏览 7提问于2017-04-28得票数 19
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
4回答
气流中的Python脚本调度
、、、、
大家好，
我需要使用气流调度我的python 文件(包含从sql和一些联接中提取的数据)。我已经成功地将气流安装到我的linux服务器上，而且我也可以使用气流的and服务器。但是，即使在阅读了文档之后，我也不清楚到底需要在哪里编写调度脚本，以及如何将该脚本应用到气流see服务器中，这样我就可以看到状态。
就配置而言，我知道dag文件夹位于我的主目录中的位置，也知道示例dags位于何处。
注意：请不要用如何在气流中运行bash脚本文件来标记这个副本，因为我需要运行位于不同位置的python。
请在气流网络服务器中找到配置如下：
下面是AIRFLOW_HOME dir中dag文件夹的屏幕
浏览 4提问于2017-01-18得票数 31
我在dags目录中有两个文件- dag_1.py和dag_2.py
dag_1.py创建静态DAG，dag_2.py根据外部json文件在某个位置创建动态DAG。
静态DAG (由dag_1.py创建)包含一个稍后阶段的任务，该任务为dag_2.py和动态DAG生成一些这些输入json文件。
过去，在不使用DAG序列化的情况下，气流1.x版本可以很好地工作。但随着气流2.0DAG系列化已成为强制性的。有时，当生成动态DAG时，我会在Scheduler中得到以下异常-
[2021-01-02 06:17:39,493] {scheduler_job.py:1293} ERROR - Except
浏览 11提问于2021-01-05得票数 4
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
