我们是否可以限制在Apache Airflow中随时运行的DAG数量
在Apache Airflow中，可以通过配置来限制同时运行的DAG数量。这个配置项称为concurrency，它定义了同时运行的DAG实例的最大数量。通过设置不同的值，可以控制Airflow并发处理任务的能力。
在Airflow的配置文件中，可以找到airflow.cfg，通过修改其中的[core]部分的parallelism参数来控制整体的并发性。这个参数决定了同时运行的任务数量，包括DAG中的任务实例。
另外，还可以通过在DAG定义中使用任务的task_concurrency属性来进一步限制DAG中各个任务的并发性。通过设置不同的值，可以确保特定任务只在有限数量的实例中运行。
限制并发的DAG数量可以帮助避免资源耗尽和性能下降的问题，尤其是在资源有限的环境中。同时，通过合理设置并发性，可以更好地控制任务的优先级和调度顺序。
以下是一些腾讯云相关产品和产品介绍链接地址，可帮助在云计算环境下使用Apache Airflow：
- 腾讯云容器服务（Tencent Kubernetes Engine，TKE）：提供弹性、高可用的容器集群管理服务，支持部署和运行Airflow容器。详情请参考：腾讯云容器服务
- 腾讯云服务器（CVM）：提供可扩展的虚拟云服务器实例，可用于搭建Airflow的运行环境。详情请参考：腾讯云服务器
- 腾讯云对象存储（Tencent Cloud Object Storage，COS）：提供高可靠、低成本的云端对象存储服务，可用于存储Airflow的日志、数据等。详情请参考：腾讯云对象存储
请注意，以上产品仅作为示例，其他云计算厂商也提供类似的产品和服务，您可以根据需求选择适合的云计算平台。
相关·内容
在1个Airflow或Cloud Composer环境中可以运行的DAG是否有最大数量？ 如果这取决于几个因素(Airflow基础设施配置、Composer集群规格、每个DAG的活动运行次数等)。影响这一点的所有因素是什么？
浏览 33提问于2020-10-13得票数 1
我们在集成slack和Cloud composer时遇到了问题。SlackAPIPostOperator具有slackclient依赖项，我们使用以下命令将其直接安装在工作节点上： sudo python3 -m pip install apache-airflow[slack] 这将安装slack 1.3.2。 我们必须在工作节点上安装，因为Cloud composer UI上的pypi包安装选项反复超时。 模块导入语句在工作节点上工作得很好。但是，composer仍会报告损坏的DAG并显示错误消息：Broken DAG: [/home/airflow/gcs/dags/composer-
浏览 17提问于2021-04-16得票数 0
我们可以限制在Apache Airflow中随时运行的DAG的数量吗？ 我们对环境中的资源是有限制的。是否有限制编号的配置。DAG在某个时间点作为一个整体在气流中运行？ 在DAG中运行的max_active_runs参数限制 有没有可能，如果一个DAG正在运行，所有其他计划的DAG都应该等待第一个DAG完成，然后按顺序触发自身？
浏览 27提问于2020-07-17得票数 1
1回答
几个小时前，这段代码运行良好，但突然之间，我的伙伴们开始陷入“排队”状态。
下面是我试图运行的内容(我只是手动触发)：
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime as dt
def test_function():
print('Hello there')
default_args = {
'owner': 'airflow',
&#
浏览 3提问于2022-05-24得票数 0
apache airflow中的##database初始化。我已经在linux上安装了apache airflow，并安装了python 3.8版本。在运行airflow命令时，它会显示版本名称。在airflow中初始化数据库时出错。解释下面的错误和解决方案，谢谢
/home/hduser/apache_airflow/venv/lib/python3.6/site-packages/sqlalchemy/orm/relationships.py:3463 SAWarning: relationship 'DagRun.serialized_dag' will copy col
浏览 2提问于2021-04-30得票数 8
1回答
我们正在简化airflow代码库的构建/部署管道。
是否有人有使用CI/CD工具为apache airflow构建和部署管道的经验？
你如何在不同的环境中部署你的airflow代码库，包括DAG/插件/操作符，比如测试，试运行，生产等。
您如何管理每个环境的airflow.cfg配置？
您在何处管理每个环境的配置。
浏览 0提问于2018-08-31得票数 6
我在运行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
回答已采纳
3回答
在我的一些Apache Airflow安装中，计划运行的DAG或任务即使在调度程序似乎没有完全加载时也不会运行。如何增加可以并发运行的DAG或任务的数量？
类似地，如果我的安装处于高负载下，并且我想限制我的Airflow工作人员拉取排队任务的速度(例如减少资源消耗)，我可以调整什么来降低平均负载？
浏览 2提问于2019-05-30得票数 45
回答已采纳
我在试着用一个空气容器。我正在使用docker桌面，并且在ubuntu终端上运行以下命令 docker run \
> -ti \
> -p 8080:8080 \
> -v C:/Users/119302/Desktop/exemplo/01_dag.py:/opt/airflow/dags/01_dag.py \
> --entrypoint=/bin/bash \
> --name airflow \
> apache/airflow:2.1.2-python3.8 \
> -c '( \
> airflow db init &a
浏览 12提问于2021-08-02得票数 0
回答已采纳
我有一个正在运行的dag，我可以使用REST API停止它吗？ 有一种方法可以删除正在运行的dag：https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html#operation/delete_dag_run。 但是没有REST API如何停止一个正在运行的dag？ 谢谢!
浏览 145提问于2021-10-09得票数 1
回答已采纳
Airflow提供rest API功能来提取dag/任务状态。 https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html#section/Trying-the-API 但是想知道是否有一种方法可以仅获得所有dag w.r.t dag所有者的最新dag/task状态，而无需为每个dag id手动指定它。 这将帮助我们为业务用户创建工作流仪表板。
浏览 27提问于2021-08-26得票数 0
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
回答已采纳
在AirFlow2.0中，您可以通过运行CLI命令airflow tasks states-for-dag-run来获取dag中任务的状态。(请参阅此处的文档：https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#state_repeat1) Airflow 1.10.12中的等价物是什么？我似乎在the 1.10.12 docs中找不到它。
浏览 66提问于2021-06-17得票数 1
回答已采纳
我有一个非常简单的测试DAG，如下所示：
from datetime import datetime
from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator
DAG = DAG(
dag_id='scheduler_test_dag',
start_date=datetime(2017, 9, 9, 4, 0, 0, 0), #..EC2 time. Equal to 11pm hora México
max_active_runs=
浏览 0提问于2018-02-13得票数 7
我使用的是airflow 1.7.1.3。
我有一个并发DAG/任务的问题。当DAG正在运行时，计划程序不再启动其他DAG。似乎调度程序完全冻结了(不再有日志) ...直到运行的DAG完成。然后，触发新的DAGrun。我的不同任务是长时间运行的ECS任务(约10分钟)
我使用了LocalExecutor，并设置了关于parallelism=32和dag_concurrency=16的默认配置。我使用airflow scheduler -n 20并自动重启它，并且为我的所有DAG声明设置'depends_on_past': False。
作为参考，我在ECS集群中运行的容器中部署
浏览 0提问于2016-10-05得票数 2
我正在尝试为我的所有dags设置两个默认值。为此，我使用以下代码在Airflow的主目录中创建了文件airflow_local_settings.py (尝试遵循https://airflow.apache.org/docs/apache-airflow/stable/concepts.html#dag-level-cluster-policy中的示例)： def dag_policy(dag: DAG):
dag.catchup = False
dag.default_args['email'] = 'blabla' 然而，我得到了一个错误：
浏览 7提问于2021-04-20得票数 1
回答已采纳
我运行了以下测试命令：
airflow test events {task_name_redacted} 2018-12-12
...and得到了以下输出：
Dependencies not met for <TaskInstance: events.{redacted} 2018-12-12T00:00:00+00:00 [None]>, dependency 'Task Instance Slots Available' FAILED: The maximum number of running tasks (16) for this task's DA
浏览 0提问于2019-01-17得票数 4
我正在考虑开始在一个项目中使用Apache airflow，并且想知道人们如何管理与Airflow的持续集成和依赖。更具体地说，我有以下设置
3 Airflow服务器: dev staging和production。
我有两个python DAG，我想把它们的源代码放在不同的仓库中。DAG本身很简单，基本上只需使用Python操作符来调用main(*args，**kwargs)。然而，由main运行的实际代码非常大，并且需要扩展几个文件/模块。每个python代码库具有不同的依赖性，例如，
Dag1使用Python2.7 pandas==0.18.1，requests=2.13.0
Dag2使
浏览 6提问于2017-07-10得票数 6
我使用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
回答已采纳
3回答
Apache气流不调度任务
、、、、
我已经安装了airflow(V1.9.0版本)和python2.7。为了测试它的安装是否正确，我尝试从浏览器中的交互式视图触发一个教程DAG。接口显示DAG正在运行，但调度程序没有显示任何活动。
下面是我尝试过的步骤
安装气流
pip install apache-airflow
安装密码用于设置使用fernet密钥的加密
pip install apache-airflow[crypto]
生成一个fernet_key并添加airflow.cfg文件：
from cryptography.fernet import Fernet
fernet_key= F
浏览 0提问于2018-02-02得票数 6
我对气流有问题，DAG中的第一项工作总是成功地开始和结束，但是第二项工作永远不会自动开始。
我试图清除UI中的作业，但是它没有启动，如果我想看到它运行，我需要删除数据库中正在运行的作业，
delete from job where state='running'
但是我没有很多处于运行状态的作业，我只有一个带有最新心跳ok的作业SchedulerJob，还有16个外部任务传感器等待这个DAG。
该池有150个插槽，有16个运行和1个计划。
我有气流调度器在运行
我让气流网络服务器在运行
在web用户界面中，所有的DAG都设置为On。
所有的DAG都有一个过去的开
浏览 1提问于2018-12-09得票数 0
回答已采纳
Airflow老兵请帮帮忙 我正在寻找cron的替代品，偶然发现了apache airflow。 我们有一个设置，其中多个用户应该能够创建他们自己的DAG并计划他们的作业。 我们的用户是可能不知道如何编写DAG python文件的混合用户。此外，他们可能无法访问运行airflow的服务器。 是否可以通过UI创建气流DAG。我找不到任何与此相同的引用。所有示例都讲述了如何创建一个python文件并将其上传到$AIRFLOW_HOME/dag/目录。用户将无权访问此目录。 例如，Rundeck允许用户通过UI添加工作流和任务依赖关系。在airflow中有没有类似的插件/功能。 PS:我真的很喜欢
浏览 155提问于2018-02-26得票数 21
回答已采纳
2回答
在Apache Airflow UI中，顶部的计时器会自动更新：
但为了跟踪特定DAG运行的进度，我必须不断单击刷新按钮：
有没有人知道是否有一种方法可以将airflow ui配置为自动更新树视图，以实时查看dag的进度？
(如果图表视图也具有相同的功能，那就更好了。)
浏览 1提问于2020-10-16得票数 2
我正在尝试使用apache airflow。我成功地安装了所有东西。
我在dag文件夹中添加了一个新的DAG，当我运行airflow list_dags时，它向我展示了与我的新dag一起的dag示例。
但是，当我进入UI时，我无法看到DAG选项卡中列出的DAG。
我已经杀了webserver并重新启动了一切。这不管用
fyi，我使用centos7在VM上运行apache。
谢谢。
浏览 0提问于2019-04-18得票数 4
回答已采纳
我试图通过运行正式的气流对接器来学习一点气流，按照这里的指令组成图像: https://airflow.apache.org/docs/apache-airflow/2.0.1/start/docker.html。我能够启动所有这样的服务，当我将我的测试DAG放在./dags文件夹中时，我最终在UI中看到了我的DAG并可以运行它。
但这花了几分钟。每次我“部署”一个新的dag到./dags时，都要花上几分钟。
我想知道是否有一些配置选项会使./dags文件夹的扫描速度更快，这样我就不用等那么久了。提前谢谢！
浏览 4提问于2022-11-30得票数 1
回答已采纳
因此，在我的例子中，我以前在我的机器上直接运行过气流，现在我尝试使用码头在容器中运行，同时保持我以前的测试记录。不过，我一直有一些问题。
一点点背景..。当我第一次使用坞-撰写来打开我的容器时，气流正在发送一条错误消息，说明列dag_has_import_errors不存在。所以我继续创造了它，一切看起来都很好。
但是，现在我的工具都坏了，当我修改一个没有解决问题的代码时，我可以在when服务器顶部显示的简短错误信息中看到更新的代码行。
但是，当我解决这个问题时，代码不会改变，DAG仍然会中断。我会提供
\
另外，下面是我的坞-撰写文件(我注释掉了气流db init，但我是否应该将它与db升
浏览 13提问于2022-02-01得票数 1
回答已采纳
1回答
我在Composer上运行Apache气流(composer-1.14.2- Airflow -1.10.14)。我想使用Terraform来创建基础设施，但是我找不到任何操作员来这样做。作为一种解决办法，我使用的BashOperator如下：
create_vm=BashOperator(
task_id='create_cluster',
bash_command=f'''
sudo apt-get update -y && \
sudo apt-get install software
浏览 1提问于2021-02-15得票数 4
我试图通过一个python脚本在mysql中自动创建一个临时表，这个脚本集成在我的气流DAG (docker)上。我使用的脚本是通过一个外部python文件使用的，该文件将通过bashcommand运行。
当我试图在气流UI上运行它时，我会得到这个错误消息。
ModuleNotFoundError:没有名为“pymysql”的模块
达格：
from airflow import DAG
from airflow.operators bash import BashOperator
from datetime import datetime, timedelta
import pymysql
浏览 4提问于2022-01-18得票数 0
在运行DAG时，在任务日志中接收以下错误：
FileNotFoundError: Errno 2没有这样的文件或目录：“直线”：“直线”
这是我的DAG：
import airflow
from airflow import DAG
from airflow.providers.apache.hive.operators.hive import HiveOperator
from airflow.utils.dates import days_ago
from datetime import timedelta
default_args = {
'owner':
浏览 3提问于2021-10-28得票数 3
回答已采纳
2回答
我想公开一个前端用户界面，以接受来自用户的一些输入参数，用于后端的数据管道，然后在前端用户界面上再次向用户显示输出。我可以让数据管道作为一个气流DAG，每当用户提交输入参数时，它就会按需运行。本质上，我想让Airflow作为web应用的后端。我可以有一个连续运行的DAG，为用户提供网页，每当他们提交输入参数时，数据流水线就会执行并显示结果吗？ 我找到了一个很接近的例子- https://towardsdatascience.com/10-minutes-to-building-a-machine-learning-pipeline-with-apache-airflow-53cd092689
浏览 12提问于2020-09-21得票数 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
我使用的是apache-气流1.10.10。
我的用例是:我有4个DAG，它们都在工作之后使用BashOperator使用不同的conf参数触发一个名为BashOperator的通用DAG。
airflow trigger_dag -c {"id":"1"} dag_common
airflow trigger_dag -c {"id":"2"} dag_common
airflow trigger_dag -c {"id":"3"} dag_common
airflow trigger_dag -
浏览 3提问于2020-06-18得票数 0
回答已采纳
我正在创建一个数据管道，通过BigQuery运算符或谷歌云库从Bigquery获取数据。但是我总是得到一个错误。下面是大查询运算符的dag：
from airflow import DAG
from datetime import datetime, timedelta
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
from airflow.contrib.operators.bigquery_check_operator import BigQueryCheckOperator
from r
浏览 13提问于2021-05-13得票数 0
在visual代码中运行气流DAG代码时出错。
误差
ImportError: cannot import name 'DummyOperator' from 'airflow.operators' (c:\Users\10679196\AppData\Local\Programs\Python\Python38\lib\site-packages\airflow\operators\__init__.py)
导入报表
from airflow import DAG
from airflow.operators import DummyOperator
版本
a
浏览 6提问于2022-11-07得票数 0
回答已采纳
3回答
我正在为我的ETL编写一个单元测试，作为一个过程，我想测试所有Dags，以确保它们没有周期。在Bas和Julian用Apache气流读取数据管道之后，我看到他们使用的是DAG.test_cycle()，这里的数据管道是从模块airflow.models.dag导入的，但是当我运行代码时，会得到一个错误：' DAG‘对象没有属性'test_cycle’。
这是我的代码片段
import glob
import importlib
import os
import pytest
from airflow.models.dag import DAG
DAG_PATH = os.p
浏览 4提问于2021-05-27得票数 3
回答已采纳
我有一个Cloud Composer环境，版本为composer-1.17.1-airflow-2.1.2。在这个环境中，我目前有27个DAG。
但是，当我转到环境的Monitoring页面，并单击"Show DAG Metrics“时，"DAG bag size”的图表显示大小为137。
是否有理由担心它不等于DAG的数量？它实际上能反映DAG运行的总次数，或者可能反映操作员的数量吗？是什么影响了该指标的大小，我如何知道该指标是否健康？
浏览 25提问于2021-09-30得票数 0
我遇到了一个问题，我可以从airflow的rest api命令(https://airflow.apache.org/api.html)成功触发dag；但是，dag实例不能运行。我正在调用-> POST /api/experimental/dags/dag_id/dag_runs，其中dag_id是我正在运行的dag。唯一发生的事情是dag立即返回成功。我手动触发了dag，并开始运行dag实例(见图2 dag运行)。注意:第二次DAG运行失败-这应该不会影响我试图解决的问题。 DAG 修复了->必须处理调度程序的问题。我添加了'depends_on_past'：F
浏览 17提问于2019-10-03得票数 0
回答已采纳
我有一个包含数万个任务的Apache Airflow DAG，在运行之后，我发现其中有几个任务失败了。 我修复了导致某些任务失败的错误，我只想重新运行失败的任务。这篇SO帖子建议使用图形用户界面“清除”失败的任务： How to restart a failed task on Airflow 如果您有少量失败的任务，则此方法有效。 我想知道我们是否可以绕过GUI，通过命令行来做有问题的事情，比如： airflow_clear_failed_tasks dag_id execution_data
浏览 50提问于2018-12-27得票数 11
回答已采纳
我想为conn_id生成spark_default。我正在k8s上运行我的k8s，我想使用火花主程序动态地生成conn_id，这是运行在同一个名称空间中的另一个容器。
是否有办法动态生成conn_id :类似于：
环境变量
或者使用SparkSubmitOperator本身编写和生成conn_id
下面是我的dag代码：
from airflow import DAG
from airflow.contrib.operators.spark_submit_operator import SparkSubmitOperator
from datetime import date
浏览 1提问于2019-05-23得票数 0
我正在探索Airflow RBAC功能(https://airflow.apache.org/docs/stable/security.html?highlight=ldap#rbac-ui-security)，但令人失望的是，它们只适用于Airflow UI，就我所知，无法确定用户可以通过the experimental API或Airflow CLI执行哪些操作。例如，我可以限制某人通过UI触发DAG，但他们仍然可以使用实验API或CLI执行此操作。 我来这里只是为了验证确实如此，并询问是否有任何其他机制来管理实验性API或CLI的使用？我怀疑答案是“没有机制来管理实验性API或CLI的
浏览 27提问于2020-07-06得票数 2
回答已采纳
在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
我正在使用apache airflow和BashOperator。每个BashOperator执行一个python脚本。例如：
from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator
with DAG(dag_id='dag_example', default_args=None,
schedule_interval='0 2 */2 * *',
catchup=False) as dag:
浏览 18提问于2020-05-20得票数 1
我试图在MWAA中使用类EcsCreateClusterOperator，但是我得到了以下错误：
Broken DAG: [/usr/local/airflow/dags/etl_basic_ecs/etl_ecs_basic_dag.py] Traceback (most recent call last):
File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
File "/usr/local/airflow/dags/etl_basic_ecs/
浏览 7提问于2022-09-05得票数 0
我最近把气流从1.10.15升级到2.2.4。使用sudo安装。升级后升级DB。气流正在运行，但也有导入错误。
我有多个带有行from airflow.operators.mssql_operator import MsSqlOperator的DAG，但是一旦调度程序和and服务器启动，我将得到以下错误：
Broken DAG: [/opt/airflow/dags/admissions.py] Traceback (most recent call last):
File "/opt/airflow/dags/admissions.py", line 7, in <
浏览 16提问于2022-03-28得票数 0
回答已采纳
我正在使用Airflow Version 2.2.5/Composer 2.0.15实现任务超时错误。在Airflow version2.2.3 /Composer Version 1.18.0中，相同的代码运行得非常好。
错误信息：
Broken DAG: [/home/airflow/gcs/dags/test_dag.py] Traceback (most recent call last):
File "/opt/python3.8/lib/python3.8/enum.py", line 256, in __new__
if canonical_membe
浏览 32提问于2022-06-13得票数 0
回答已采纳
2回答
我正在尝试使用python代码中的模板并按照定义的使用globals()来生成airflow dags
定义dag对象并保存。下面是我的代码：
import datetime as dt
import sys
import airflow
from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator
argumentList = sys.argv
owner = argumentList[1]
dag_name = argumentList[2]
taskID = argum
浏览 9提问于2020-06-04得票数 0
我正在使用Airflow EMR操作符创建一个AWS EMR集群，该集群运行包含在S3中的Jar文件，然后将输出写回S3。它似乎能够使用S3中的Jar文件运行作业，但我无法让它将输出写入S3。当作为AWS EMR CLI Bash命令运行它时，我能够让它将输出写入S3，但我需要使用Airflow EMR操作符来执行此操作。我在Airflow步骤配置和Jar文件的环境配置中都设置了S3输出目录，但仍然无法让操作员对其进行写入。 这是我的Airflow DAG的代码 from datetime import datetime, timedelta
import airflow
from airf
浏览 13提问于2019-09-12得票数 0
回答已采纳
我对使用apache airflow非常陌生。我使用pycharm作为我的IDE。我创建了一个项目(anaconda环境)，创建了一个包含DAG定义和Bash操作符的python脚本。当我打开airflow When服务器时，我的DAGS不显示。仅显示默认示例DAG。我的AIRFLOW_HOME变量包含~/airflow。所以我在那里存储了我的python脚本，现在它显示出来了。
如何在项目环境中使用它？
我是否要在每个项目开始时更改环境变量？
有没有办法为每个项目添加特定的airflow主目录？
我不想将我的DAG存储在默认的airflow目录中，因为我想将它添加到我的git库中。请帮帮我。
浏览 2提问于2019-07-04得票数 7
回答已采纳
当我在终端中运行它时，下面的代码运行正常：
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
回答已采纳
我已经在一个码头上安装了apache_beam Python和airflow。Python版本: 3.5
阿帕奇气流: 1.10.5
我正在尝试使用**DataflowPythonOperator**执行apache波束管道。当我从气流用户界面运行DAG时，我得到
导入错误:导入apache_beam作为光束。模块未找到
在相同的设置下，我尝试了**DataflowTemplateOperator**，它运行得非常好。
当我尝试使用Python2和airflow 1.10.3进行相同的对接设置时，两个月前，操作符没有返回任何错误，而是按预期工作。
在SSH进入码头后，当我在一个码头容器
浏览 1提问于2019-09-06得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
