展开

关键词

Airflow速用

/concepts.html#bitshift-composition 提高airflow相关执行速度方法 通过修改airflow.cfg相关配置 官方文档如下:http://airflow.apache.org AIRFLOW_HOME="/mnt/e/project/airflow_config/local" 命令行:pip install apache-airflow 根据airflow.cfg的数据库配置 ,在连接的数据库服务创建一个 名为 airflow_db的数据库 命令行初始化数据库:airflow initdb 命令行启动web服务: airflow webserver -p 8080 启动及关闭airflow内置 dag示例方法(能够快速学习Airflow)  开启:修改airflow.cfg配置文件  load_examples = True  并重启即可  关闭:修改airflow.cfg Airflow has a shortcut to start 398 # it `airflow flower`.

1.7K10

大数据调度平台Airflow(一):什么是Airflow

什么是Airflow Apache Airflow是一个提供基于DAG有向无环图来编排工作流的、可视化的分布式任务调度平台,与Oozie、Azkaban等任务流调度平台类似。 Airflow在2014年由Airbnb发起,2016年3月进入Apache基金会,在2019年1月成为顶级项目。 另外,Airflow提供了WebUI可视化界面,提供了工作流节点的运行监控,可以查看每个节点的运行状态、运行耗时、执行日志等。 在Airflow中工作流上每个task都是原子可重试的,一个工作流某个环节的task失败可自动或手动进行重试,不必从头开始跑。 Airflow官网:http://airflow.apache.org/,Airflow支持的任务调度类型如下:

28231
  • 广告
    关闭

    【玩转 Cloud Studio】有奖调研征文,千元豪礼等你拿!

    想听听你玩转的独门秘籍,更有机械键盘、鹅厂公仔、CODING 定制公仔等你来拿!

  • 您找到你想要的搜索结果了吗?
    是的
    没有找到

    大数据调度平台Airflow(五):Airflow使用

    Airflow包。 /docs/apache-airflow/stable/_api/airflow/models/baseoperator/index.html#module-airflow.models.baseoperator 图片DAG参数说明可以参照:http://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/models/dag/index.html 6、重启Airflow“ps aux|grep webserver”和“ps aux|grep scheduler”找到对应的airflow进程杀掉,重新启动Airflow。 /dags下,重启airflow,DAG执行调度如下:图片有两种方式在Airflow中配置catchup:全局配置在airflow配置文件airflow.cfg的scheduler部分下,设置catchup_by_default

    76731

    大数据调度平台Airflow(三):Airflow单机搭建

    、启动Airflow 1、启动webserver 2、启动scheduler 3、访问Airflow webui Airflow单机搭建 Airflow是基于Python的,就是Python中的一个包。 单节点部署airflow时,所有airflow 进程都运行在一台机器上,架构图如下: 1、安装Airflow必须需要的系统依赖 Airflow正常使用必须需要一些系统依赖,在mynode4节点上安装以下依赖 Airflow文件存储目录默认在/root/airflow目录下,但是这个目录需要执行下“airflow version”后自动创建,查看安装Airflow版本信息: (python37) [root@ 这样安装完成的airflow后,查看对应的版本会将“AIRFLOW_HOME”配置的目录当做airflow的文件存储目录。 4、配置Airflow使用的数据库为MySQL 打开配置的airflow文件存储目录,默认在$AIRFLOW_HOME目录“/root/airflow”中,会有“airflow.cfg”配置文件,修改配置如下

    48631

    闲聊Airflow 2.0

    在 2020 年 12 月 17 日 Apache Airflow 团队发布了 Apache Airflow 2.0.0。 当时就想写写 Airflow 的新特性,但是粗略的看了下《Apache Airflow 2.0 is here!》 等了半年后,注意到 Airflow 已经发布版本到 2.1.1 了,而且Airflow 1.0+的版本也即将不再维护,自己也做了小规模测试,基本上可以确定 Airflow2.0 可以作为生产环境下的版本了 在Airflow 2.0中,已根据可与Airflow一起使用的外部系统对模块进行了重组。 /apache-airflow-2-0-tutorial-41329bbf7211 https://airflow.apache.org/blog/airflow-two-point-oh-is-here

    63630

    Airflow安装教程

    安装airflow [root@node1 ~]# pip install airflow 如果上面命令安装较慢,可以使用下面命令国内源安装。 [root@node1 ~]# pip install -i https://pypi.tuna.tsinghua.edu.cn/simple airflow 3.初始化数据库 airflow默认使用sqlite 作为数据库, 直接执行数据库初始化命令后, 会在环境变量路径下新建一个数据库文件airflow.db [root@node1 ~]# airflow initdb [2017-10-06 10:10:45,462 ] {__init__.py:57} INFO - Using executor SequentialExecutor DB: sqlite:////root/airflow/airflow.db [2017 启动airflow webserver 默认端口为8080 [root@node1 ~]# airflow webserver [2017-10-06 10:11:37,313] {__init__.py

    2.3K70

    大数据调度平台Airflow(六):Airflow Operators及案例

    及调度HQL 四、​​​​​​​PythonOperator Airflow Operators及案例 Airflow中最重要的还是各种Operator,其允许生成特定类型的任务,这个任务在实例化时称为 关于BaseOperator的参数可以参照: http://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/models/baseoperator 在default_args中的email是指当DAG执行失败时,发送邮件到指定邮箱,想要使用airflow发送邮件,需要在$AIRFLOW_HOME/airflow.cfg中配置如下内容: [smtp] ) [root@node4 ~]# airflow webserver --port 8080 (python37) [root@node4 ~]# airflow scheduler 2、配置SSH ==2.0.2 #启动airflow (python37) [root@node4 ~]# airflow webserver --port 8080 (python37) [root@node4 ~

    54051

    Apache AirFlow 入门

    Airflow是一个可编程,调度和监控的工作流平台,基于有向无环图(DAG),airflow可以定义一组有依赖的任务,按照依赖依次执行。 官方网站-AirFlow AirFlow-中文文档 定义 Pipeline 导入模块 一个 Airflow 的 pipeline 就是一个 Python 脚本,这个脚本的作用是为了定义 Airflow # DAG 对象; 我们将需要它来实例化一个 DAG from airflow import DAG # Operators 我们需要利用这个对象去执行流程 from airflow.operators.bash 此时,您的代码应如下所示: """ Airflow 教程代码位于: https://github.com/apache/airflow/blob/master/airflow/example_dags /tutorial.py """ from airflow import DAG from airflow.operators.bash_operator import BashOperator from

    44300

    airflow 配置 CeleryExecutor

    本文介绍如何配置 airflow 的 CeleryExecutor。 操作步骤 CeleryExecutor 需要 Python 环境安装有 celery。 /redis-server redis.conf 2>1& 第三步:配置 airflow.cfg 修改 airflow.cfg #修改 3 处: executor = CeleryExecutor broker_url #启动webserver #后台运行 airflow webserver -p 8080 -D airflow webserver -p 8080 #启动scheduler #后台运行 airflow scheduler -D airflow scheduler #启动worker #后台运行 airflow worker -D #如提示addres already use ,则查看 worker_log_server_port = 8793 是否被占用,如是则修改为 8974 等 #未被占用的端口 airflow worker #启动flower -- 可以不启动 #后台运行 airflow flower -D airflow

    92820

    大数据调度平台Airflow(四):Airflow WebUI操作介绍

    Airflow WebUI操作介绍 一、DAG DAG有对应的id,其id全局唯一,DAG是airflow的核心概念,任务装载到DAG中,封装成任务依赖链条,DAG决定这些任务的执行规则。 二、​​​​​​​Security “Security”涉及到Airflow中用户、用户角色、用户状态、权限等配置。 三、​​​​​​​Browse DAG Runs 显示所有DAG状态 Jobs  显示Airflow中运行的DAG任务 Audit Logs 审计日志,查看所有DAG下面对应的task的日志,并且包含检索 四、​​​​​​​Admin 在Admin标签下可以定义Airflow变量、配置Airflow、配置外部连接等。 五、​​​​​​​Docs Docs中是关于用户使用Airflow的一些官方使用说明文档连接。

    19131

    Airflow Dag可视化管理编辑工具Airflow Console

    Airflow Console: https://github.com/Ryan-Miao/airflow-console Apache Airflow扩展组件, 可以辅助生成dag, 并存储到git 如何使用 一些概念 DAG: Airflow原生的dag, 多个任务依赖组成的有向无环图, 一个任务依赖链。 Airflow那边定时拉取git更新即可. ? 本地启动 通过docker-airflow 启动airflow, 暴露pg端口和webserver端口, docker-compose.yml cd doc docker-compose up 启动后访问 localhost:8090即airflow初始化完成.

    1.3K30

    大数据调度平台Airflow(二):Airflow架构及原理

    目录 Airflow架构及原理 一、Airflow架构 二、Airflow术语 三、Airflow工作原理 Airflow架构及原理 一、Airflow架构 Airflow我们可以构建Workflow工作流 ,工作流使用DAG有向无环图来表示,DAG指定了任务之间的关系,如下图: Airflow架构图如下: Airflow在运行时有很多守护进程,这些进程提供了airflow全部功能,守护进程包括如下: DaskExecutor:动态任务调度,支持远程集群执行airflow任务。 关于不同Executor类型可以参考官网:https://airflow.apache.org/docs/apache-airflow/stable/executor/index.html work: 三、​​​​​​​Airflow工作原理 airflow中各个进程彼此之间是独立不互相依赖,也不互相感知,每个进程在运行时只处理分配到自身的任务,各个进程在一起运行,提供了Airflow全部功能,其工作原理如下

    52031

    Airflow配置和使用

    Airflow能做什么 Airflow是一个工作流分配管理系统,通过有向非循环图的方式管理任务流程,设置任务依赖关系和时间调度。 Airflow独立于我们要运行的任务,只需要把任务的名字和运行方式提供给Airflow作为一个task就可以。 airflow: airflow initdb` (若前面执行过,就跳过) ct@server:~/airflow: airflow webserver --debug & ct@server:~/airflow /usr/local/bin/airflow worker user=airflow environment=AIRFLOW_HOME="/home/airflow/airflow", PATH="/ airflow environment=AIRFLOW_HOME="/home/airflow/airflow", PATH="/usr/local/bin:%(ENV_PATH)s" stderr_logfile

    10.8K71

    docker上安装airflow

    centos 7环境下: mkdir airflow  //创建airflow文件夹 git clone https://github.com/puckel/docker-airflow.git /root /airflow //下载源码到airflow文件夹 docker run -d -p 8082:8080 puckel/docker-airflow  //安装并运行airflow docker exec -it af2044c3b40c bash // 进入容器 airflow initdb  // 初始化数据库 出现错误: airflow.exceptions.AirflowException: Could 解决办法: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" export AIRFLOW airflow initdb // 重新运行初始化数据库 输入网址: http://172.16.10.22:8083/admin/,效果图如下: ?

    1.3K10

    【翻译】Airflow最佳实践

    原文:https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html 创建DAG有两个步骤: 用Python实现一个 1.1 实现自定义算子(Operator)或者钩子(Hook) 具体看这里:https://airflow.apache.org/docs/apache-airflow/stable/howto/custom-operator.html 关于Connection:https://airflow.apache.org/docs/apache-airflow/stable/concepts/connections.html 1.5 变量Variables 在Airflow中,使用变量去连接到元数据DB,获取数据,这会减慢解释的速度,并给数据库增加额外的负担。 对于变量,使用AIRFLOW_VAR_{KEY}: with mock.patch.dict('os.environ', AIRFLOW_VAR_KEY="env-value"): assert

    32910

    Airflow 和 DataX 的结合

    我们团队用的调度系统是 Apache Airflow(https://github.com/apache/airflow),数据传输工具是 DataX(https://github.com/alibaba Apache Airflow 自身也带了一些数据传输的 Operator ,比如这里的https://github.com/apache/airflow/blob/main/airflow/operators 网上也有一些文章讲如何将 Airflow 和 DataX 结合起来,比如有: https://www.cnblogs.com/woshimrf/p/airflow-plugin.html https:/ /tech.youzan.com/data_platform/ 对于文章 1,虽然结合了 Airflow 和 DataX,但是它并没有解决 Airflow 的网状链路问题,只是用 Airflow 调用 Hive 里对应的的表名和 Airflow 的 connection id,最后再补充下定时调度的相关配置信息,就完成了一次数据传输的开发。

    50920

    实用调度工具Airflow

    Airflow 这里介绍一个Airflow,这个是由Airbnb公司贡献的,(Airbnb,是一个让大众出租住宿民宿的网站,提供短期出租房屋或房间的服务。最近业务也开到中国来了) 。 """ Code that goes along with the Airflow tutorial located at: https://github.com/airbnb/airflow/blob /master/airflow/example_dags/tutorial.py """ from airflow import DAG from airflow.operators.bash_operator ': False, 'start_date': datetime(2015, 6, 1), 'email': ['airflow@airflow.com'], 'email_on_failure': False http://airflow.incubator.apache.org/profiling.html 4 扩展性方面支持和Celery和mesos集成 5 最后再看看社区状况,人不少,281个 ?

    2.7K60

    airflow —安装详解 (2)

    安装python>=2.7 安装airflow 安装airflow :https://pythonhosted.org/airflow/installation.html yum install libxml2 pip2 install airflow[hive] pip2 install airflow[mysql] pip2 install airflow[celery] pip2 install airflow flower 安装DB # initialize the database airflow initdb 如果需要mysql,请修改全局配置文件airflow.cfg,并创建airflow库: create flower 查看配置文件 $ cd ~/airflow $ ls airflow.cfg airflow.db airflow-webserver.pid unittests.cfg 开启用户认证 创建用户 import airflow from airflow import models, settings from airflow.contrib.auth.backends.password_auth

    4.1K10

    认识Airflow的DAG

    前文Airflow的第一个DAG已经跑起来了我们的第一个任务. 本文就来丰富这个任务. 回顾我们的任务内容 ? 目前,内置了一些: https://github.com/apache/airflow/tree/master/airflow/operators 第三方也贡献了一些: https://github.com /apache/airflow/tree/master/airflow/contrib/operators 还可以自己编写plugin, 制作自己的任务类型插件. 当想要使用这些插件的时候,只要引入 from airflow.operators.bash_operator import BashOperator from airflow.operators.dummy_operator 所以,Airflow提供了通知回调。

    97340

    大数据调度平台Airflow(七):Airflow分布式集群搭建原因及其他扩展

    目录 Airflow分布式集群搭建原因及其他扩展 一、Airflow分布式集群搭建原因 二、Airflow分布式集群其他扩展 1、​​​​​​​扩展Worker节点 2、​​​​​​​扩展Master节点 3、​​​​​​​Scheduler HA Airflow分布式集群搭建原因及其他扩展 一、Airflow分布式集群搭建原因 在稳定性要求较高的场景中,例如:金融交易系统,airflow一般采用集群、 高可用方式搭建部署,airflow对应的进程分布在多个节点上运行,形成Airflow集群、高可用部署,架构图如下: 以上集群、高可用方式搭建Airflow好处如下: 如果一个worker节点崩溃挂掉, 我们也可以通过增加单个worker节点的进程数来垂直扩展集群,可以通过修改airflow配置文件AIRFLOW_HOME/airflow.cfg中celeryd_concurrency的值来实现,例如: Master扩展参照后续Airflow分布式集群搭建,扩展Master后的架构如下: 3、​​​​​​​Scheduler HA 扩展Master后的Airflow集群中只能运行一个Scheduler

    30951

    相关产品

    • 腾讯乐享

      腾讯乐享

      腾讯乐享,凝聚腾讯十年管理经验与精华,通过直播、文档、课堂、考试、学习地图、论坛、乐问、投票、活动、祝福、积分和证书等核心应用,满足政府、企业和学校在知识管理、学习培训、文化建设和内部沟通的多元化需求,提供全行业、多场景解决方案,助力数字化管理升级。

    相关资讯

    热门标签

    活动推荐

    扫码关注腾讯云开发者

    领取腾讯云代金券