助力工业物联网,工业大数据之服务域:AirFlow的架构组件【三十二】

本文涉及的产品
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
简介: 助力工业物联网,工业大数据之服务域:AirFlow的架构组件【三十二】

知识点05:AirFlow的架构组件

  • 目标:了解AirFlow的架构组件
  • 路径
  • step1:架构
  • step2:组件
  • 实施
  • 架构
  • Client:开发AirFlow调度的程序的客户端,用于开发AirFlow的Python程序
  • Master:分布式架构中的主节点,负责运行WebServer和Scheduler
  • Worker:负责运行Execution执行提交的工作流中的Task
  • 组件
A scheduler, which handles both triggering scheduled workflows, and submitting Tasks to the executor to run.
An executor, which handles running tasks. In the default Airflow installation, this runs everything inside the scheduler, but most production-suitable executors actually push task execution out to workers.
A webserver, which presents a handy user interface to inspect, trigger and debug the behaviour of DAGs and tasks.
A folder of DAG files, read by the scheduler and executor (and any workers the executor has)
A metadata database, used by the scheduler, executor and webserver to store state.
  • WebServer:提供交互界面和监控,让开发者调试和监控所有Task的运行
  • Scheduler:负责解析和调度Task任务提交到Execution中运行
  • Executor:执行组件,负责运行Scheduler分配的Task,运行在Worker中
  • DAG Directory:DAG程序的目录,将自己开发的程序放入这个目录,AirFlow的WebServer和Scheduler会自动读取
  • airflow将所有程序放在一个目录中
  • 自动检测这个目录有么有新的程序
  • MetaData DataBase:AirFlow的元数据存储数据库,记录所有DAG程序的信息
  • 小结
  • 了解AirFlow的架构组件

知识点06:AirFlow的开发规则

  • 目标掌握AirFlow的开发规则
  • 路径
  • step1:开发Python调度程序
  • step2:提交Python调度程序
  • 实施
  • 官方文档
  • 示例:http://airflow.apache.org/docs/apache-airflow/stable/tutorial.html
  • 开发Python调度程序
  • 开发一个Python程序,程序文件中需要包含以下几个部分
  • 注意:该文件的运行不支持utf8编码,不能写中文
  • step1:导包
# 必选:导入airflow的DAG工作流
from airflow import DAG
# 必选:导入具体的TaskOperator类型
from airflow.operators.bash import BashOperator
# 可选:导入定时工具的包
from airflow.utils.dates import days_ago
  • step2:定义DAG及配置
# 当前工作流的基础配置
default_args = {
    # 当前工作流的所有者
    'owner': 'airflow',
    # 当前工作流的邮件接受者邮箱
    'email': ['airflow@example.com'],
    # 工作流失败是否发送邮件告警
    'email_on_failure': True,
    # 工作流重试是否发送邮件告警
    'email_on_retry': True,
    # 重试次数
    'retries': 2,
    # 重试间隔时间
    'retry_delay': timedelta(minutes=1),
}
# 定义当前工作流的DAG对象
dagName = DAG(
    # 当前工作流的名称,唯一id
    'airflow_name',
    # 使用的参数配置
    default_args=default_args,
    # 当前工作流的描述
    description='first airflow task DAG',
    # 当前工作流的调度周期:定时调度【可选】
    schedule_interval=timedelta(days=1),
    # 工作流开始调度的时间
    start_date=days_ago(1),
    # 当前工作流属于哪个组
    tags=['itcast_bash'],
)
  • 构建一个DAG工作流的实例和配置
  • step3:定义Tasks
  • 执行Linux命令
  • 执行Python代码
  • 发送邮件的
  • 其他
  • BashOperator:定义一个Shell命令的Task
# 导入BashOperator
from airflow.operators.bash import BashOperator
# 定义一个Task的对象
t1 = BashOperator(
  # 指定唯一的Task的名称
    task_id='first_bashoperator_task',
  # 指定具体要执行的Linux命令
    bash_command='echo "hello airflow"',
  # 指定属于哪个DAG对象
    dag=dagName
)
  • PythonOperator:定义一个Python代码的Task
# 导入PythonOperator
from airflow.operators.python import PythonOperator
# 定义需要执行的代码逻辑
def sayHello():
    print("this is a programe")
#定义一个Task对象
t2 = PythonOperator(
    # 指定唯一的Task的名称
    task_id='first_pyoperator_task',
    # 指定调用哪个Python函数
    python_callable=sayHello,
    # 指定属于哪个DAG对象
    dag=dagName
)
  • step4:运行Task并指定依赖关系
  • 定义Task
Task1:runme_0
Task2:runme_1
Task3:runme_2
Task4:run_after_loop
Task5:also_run_this
Task6:this_will_skip
Task7:run_this_last
  • 需求
  • Task1、Task2、Task3并行运行,结束以后运行Task4
  • Task4、Task5、Task6并行运行,结束以后运行Task7

  • 代码
task1 >> task4
task2 >> task4
task3 >> task4
task4 >> task7
task5 >> task7
task6 >> task7
  • 如果只有一个Task,只要直接写上Task对象名称即可
task1
  • 提交Python调度程序
  • 哪种提交都需要等待一段时间
  • 自动提交:需要等待自动检测
  • 将开发好的程序放入AirFlow的DAG Directory目录中
  • 默认路径为:/root/airflow/dags
  • 手动提交:手动运行文件让airflow监听加载
python xxxx.py
  • 调度状态
  • No status (scheduler created empty task instance):调度任务已创建,还未产生任务实例
  • Scheduled (scheduler determined task instance needs to run):调度任务已生成任务实例,待运行
  • Queued (scheduler sent task to executor to run on the queue):调度任务开始在executor执行前,在队列中
  • Running (worker picked up a task and is now running it):任务在worker节点上执行中
  • Success (task completed):任务执行成功完成
  • 小结
  • 掌握AirFlow的开发规则


相关实践学习
钉钉群中如何接收IoT温控器数据告警通知
本实验主要介绍如何将温控器设备以MQTT协议接入IoT物联网平台,通过云产品流转到函数计算FC,调用钉钉群机器人API,实时推送温湿度消息到钉钉群。
阿里云AIoT物联网开发实战
本课程将由物联网专家带你熟悉阿里云AIoT物联网领域全套云产品,7天轻松搭建基于Arduino的端到端物联网场景应用。 开始学习前,请先开通下方两个云产品,让学习更流畅: IoT物联网平台:https://iot.console.aliyun.com/ LinkWAN物联网络管理平台:https://linkwan.console.aliyun.com/service-open
目录
相关文章
|
8天前
|
大数据
【赵渝强老师】大数据主从架构的单点故障
大数据体系架构中,核心组件采用主从架构,存在单点故障问题。为提高系统可用性,需实现高可用(HA)架构,通常借助ZooKeeper来实现。ZooKeeper提供配置维护、分布式同步等功能,确保集群稳定运行。下图展示了基于ZooKeeper的HDFS HA架构。
|
1月前
|
SQL 存储 分布式计算
ODPS技术架构深度剖析与实战指南——从零开始掌握阿里巴巴大数据处理平台的核心要义与应用技巧
【10月更文挑战第9天】ODPS是阿里巴巴推出的大数据处理平台,支持海量数据的存储与计算,适用于数据仓库、数据挖掘等场景。其核心组件涵盖数据存储、计算引擎、任务调度、资源管理和用户界面,确保数据处理的稳定、安全与高效。通过创建项目、上传数据、编写SQL或MapReduce程序,用户可轻松完成复杂的数据处理任务。示例展示了如何使用ODPS SQL查询每个用户的最早登录时间。
91 1
|
1月前
|
存储 分布式计算 大数据
大数据-169 Elasticsearch 索引使用 与 架构概念 增删改查
大数据-169 Elasticsearch 索引使用 与 架构概念 增删改查
57 3
|
9天前
|
SQL 数据采集 分布式计算
【赵渝强老师】基于大数据组件的平台架构
本文介绍了大数据平台的总体架构及各层的功能。大数据平台架构分为五层:数据源层、数据采集层、大数据平台层、数据仓库层和应用层。其中,大数据平台层为核心,负责数据的存储和计算,支持离线和实时数据处理。数据仓库层则基于大数据平台构建数据模型,应用层则利用这些模型实现具体的应用场景。文中还提供了Lambda和Kappa架构的视频讲解。
【赵渝强老师】基于大数据组件的平台架构
|
29天前
|
消息中间件 Kafka 数据库
微服务架构中,如何确保服务之间的数据一致性?
微服务架构中,如何确保服务之间的数据一致性?
|
8天前
|
SQL 分布式计算 大数据
【赵渝强老师】大数据生态圈中的组件
本文介绍了大数据体系架构中的主要组件,包括Hadoop、Spark和Flink生态圈中的数据存储、计算和分析组件。数据存储组件包括HDFS、HBase、Hive和Kafka;计算组件包括MapReduce、Spark Core、Flink DataSet、Spark Streaming和Flink DataStream;分析组件包括Hive、Spark SQL和Flink SQL。文中还提供了相关组件的详细介绍和视频讲解。
|
1月前
|
存储 SQL 分布式计算
湖仓一体架构深度解析:构建企业级数据管理与分析的新基石
【10月更文挑战第7天】湖仓一体架构深度解析:构建企业级数据管理与分析的新基石
68 1
|
1月前
|
存储 SQL 缓存
Apache Doris 3.0 里程碑版本|存算分离架构升级、湖仓一体再进化
从 3.0 系列版本开始,Apache Doris 开始支持存算分离模式,用户可以在集群部署时选择采用存算一体模式或存算分离模式。基于云原生存算分离的架构,用户可以通过多计算集群实现查询负载间的物理隔离以及读写负载隔离,并借助对象存储或 HDFS 等低成本的共享存储系统来大幅降低存储成本。
Apache Doris 3.0 里程碑版本|存算分离架构升级、湖仓一体再进化
|
1月前
|
存储 机器学习/深度学习 分布式计算
大数据技术——解锁数据的力量,引领未来趋势
【10月更文挑战第5天】大数据技术——解锁数据的力量,引领未来趋势
|
5天前
|
存储 分布式计算 数据挖掘
数据架构 ODPS 是什么?
数据架构 ODPS 是什么?
50 7

热门文章

最新文章