airflow

1热度

1回答

dag = DAG( dag_id='error_config', default_args=default_args, schedule_interval= '*/10 * * * *') 我DAG的ARG游戏这样的，但是当我点击了网页上的“运行”按钮，此DAG只有正常运行，但未来schedule_interval后不能自动运行

5热度

1回答

气流DAG运行已触发，但从未执行？

我发现自己处于手动触发DAG运行（通过airflow trigger_dag datablocks_dag）运行的情况，并且Dag Run在界面中显示，但它会永远保持“运行”状态，而无需执行任何操作。当我在UI检查该DAG运行，我看到以下内容：我有start_date设置为datetime(2016, 1, 1)，并schedule_interval设置为@once。我的从阅读文档的理解是，

2热度

1回答

气流相对导入外部/ dag目录

我无法将通用代码移到气流使用的dag目录之外。我查看了airflow source，发现imp.load_source。是否可以使用imp.load_source来加载dag目录之外的模块？在下面的例子中，这将从公共目录导入foo或bar。 ── airflow_home |──── dags │ ├── dag_1.py │ └── dag_2.py ├

0热度

1回答

Jinja没有在气流DAG中转换变量

我有一个Airflow实例不会转换Jinja变量。我在我的DAG中有这个代码。 print 'execution date is: ', ' {{ ds }} ' 上述变量是字面打印的。 INFO - Subtask: execution date is: {{ ds }} Jinja2通过pip安装（我使用Python 2.7）。 PYTHONPATH设置正确，因为来自jinja模块的i

0热度

1回答

如何将参数提交给Airflow中的Python程序？

我们有一个直接与DAG API（DagBag()，get_dag()然后dag_run()）的DAG运行正常，问题是，我们能不能找到一种方法与执行这样的DAG执行的DAG前端服务器具体参数。最接近的解决方案是使用变量API，它使用set()和get()方法，但这些变量是全局变量，并且在使用相同变量名称的并发操作中工作时可能会发生冲突。我们如何运行一个dag并设置可用的参数来执行它？我们主要使

0热度

1回答

动态创建任务列表

我有一个DAG，它是通过查询DynamoDB获取列表创建的，并且列表中的每个项目都使用PythonOperator创建并将其添加到DAG中。在下面的例子中没有显示，但重要的是要注意列表中的一些项目取决于其他任务，所以我使用set_upstream来强制执行依赖关系。 - airflow_home \- dags \- workflow.py workflow.py def

2热度

1回答

如何使用气流与芹菜

我是新来的气流和芹菜，我已经完成绘制DAG了，但我想运行任务在两台计算机是在同一子网，我想知道如何修改airflow.cfg。一些例子可能会更好。感谢任何答案orz。

1热度

1回答

在AIRFLOW中访问修改的气流变量作为S3传感器的自定义参数

让我们来看看我能否在这个问题上解释我自己。请参阅我的S3文件，我从格式错误的客户那里收到。例如，日期显示为小破折号，例如“2017_07_10”。由于我想访问它们以便能够下载它们，首先我有一个任务是气流中的S3传感器。它看起来像这样： xxx = S3KeySensor( task_id='task_name', bucket_key=BUCKET_KEY, wil

1热度

4回答

如何在退出ssh时让您的气流调度程序在AWS EC2中运行？

嗨，我正在使用Airflow并将我的气流项目放在EC2中。然而，当我的mac进入睡眠状态或退出ssh时，如何保持气流调度程序运行？

1热度

2回答

如何使用--daemon停止/检查airflow webserver/scheduler的日志？

嗨，我正在使用Airflow并在EC2的后台运行。但是，您如何停止/检查日志？我找不到文档的详细说明。谢谢！