大数据技术原理与应用

大数据技术原理与应用 http://dblab.xmu.edu.cn/post/bigdata 温馨提示 : 编辑幻灯片母版, 可以修改每页 PPT 的厦大校徽和底部文字 第十一讲流计算 (PPT 版本号 :2016 年 4 月 27 日版本 ) 林子雨厦门大学计算机科学系 E-mail: ziyulin@xmu.edu.cn 主页 :http://www.cs.xmu.edu.cn/linziyu 大数据技术原理与应用 厦门大学计算机科学系厦门大学计算机科学系林子雨 2016 年版 ziyulin@xmu.edu.cn

课堂内容与教材对应关系说明 厦门大学林子雨编著 大数据技术原理与应用 2015 年 8 月 1 日人民邮电出版社出版发行第 1 版教材共包含 13 章内容 第一章大数据概述第二章大数据处理架构 Hadoop 第三章分布式文件系统 HDFS 第四章分布式数据库 HBase 第五章 NoSQL 数据库第六章云数据库第七章 MapReduce 第八章流计算第九章图计算第十章数据可视化第十一章大数据在互联网领域的应用第十二章大数据在生物医学领域的应用 ( 自学 ) 第十三章大数据的其他应用 ( 自学 ) 2016 年新增章节 ( 将加入到第 2 版教材中 ) 第 14 章基于 Hadoop 的数据仓库 Hive 第 15 章 Hadoop 架构再探讨第 16 章 Spark

课堂内容与教材对应关系说明 课堂章节第 1 讲 - 大数据概述第 2 讲 - 大数据处理架构 Hadoop 第 3 讲 - 分布式文件系统 HDFS 第 4 讲 - 分布式数据库 HBase 第 5 讲 -NoSQL 数据库第 6 讲 - 云数据库第 7 讲 -MapReduce 第 8 讲 - 基于 Hadoop 的数据仓库 Hive 第 9 讲 -Hadoop 架构再探讨第 10 讲 -Spark 第 11 讲 - 流计算第 12 讲 - 图计算第 13 讲 - 数据可视化第 14 讲 - 大数据在互联网领域的应用 对应的 大数据技术原理与应用 ( 第 1 版 ) 教材章节 第 1 章 - 大数据概述第 2 章 - 大数据处理架构 Hadoop 第 3 章 - 分布式文件系统 HDFS 第 4 章 - 分布式数据库 HBase 第 5 章 -NoSQL 数据库第 6 章 - 云数据库第 7 章 -MapReduce 新增第 14 章, 不在当前第 1 版教材中, 将放入第 2 版教材 新增第 15 章, 不在当前第 1 版教材中, 将放入第 2 版教材新增第 16 章, 不在当前第 1 版教材中, 将放入第 2 版教材 第 8 章 - 流计算第 9 章 - 图计算第 10 章 - 数据可视化第 11 章 - 大数据在互联网领域的应用备注 : 教材的第 12 章大数据在生物医学领域的应用和第 13 章大数据在其他领域的应用, 为自学章节, 不录制视频

大数据技术原理与应用 厦门大学计算机科学系厦门大学计算机科学系林子雨 2016 年版 ziyulin@xmu.edu.cn 大数据技术原理与应用 http://dblab.xmu.edu.cn/post/bigdata 温馨提示 : 编辑幻灯片母版, 可以修改每页 PPT 的厦大校徽和底部文字 第八章流计算 (PPT 版本号 :2016 年 4 月 27 日版本 ) 林子雨厦门大学计算机科学系 E-mail: ziyulin@xmu.edu.cn 主页 :http://www.cs.xmu.edu.cn/linziyu

提纲 8.1 流计算概述 8.2 流计算处理流程 8.3 流计算应用 8.4 流计算开源框架 Storm 8.5 Spark Streaming 8.6 Samza 8.7 Storm Spark Streaming 和 Samza 的应用场景 8.8 Storm 编程实践 本 PPT 是如下教材的配套讲义 : 21 世纪高等教育计算机规划教材 大数据技术原理与应用 概念 存储 处理 分析与应用 (2015 年 8 月第 1 版 ) 厦门大学林子雨编著, 人民邮电出版社 ISBN:978-7-115-39287-9 欢迎访问 大数据技术原理与应用 教材官方网站 : http://dblab.xmu.edu.cn/post/bigdata 欢迎访问 中国高校大数据课程公共服务平台 旗下子栏目 大数据课程学生服务站, 为学生学习大数据课程提供全方位 一站式免费服务 : http://dblab.xmu.edu.cn/post/4331/

8.1 流计算概述 8.1.1 静态数据和流数据 8.1.2 批量计算和实时计算 8.1.3 流计算概念 8.1.4 流计算与 Hadoop 8.1.5 流计算框架

8.1.1 静态数据和流数据 很多企业为了支持决策分析而构建的数据仓库系统, 其中存放的大量历史数据就是静态数据 技术人员可以利用数据挖掘和 OLAP(On- Line Analytical Processing) 分析工具从静态数据中找到对企业有价值的信息

8.1.1 静态数据和流数据 近年来, 在 Web 应用 网络监控 传感监测等领域, 兴起了一种新的数据密集型应用 流数据, 即数据以大量 快速 时变的流形式持续到达 实例 :PM2.5 检测 电子商务网站用户点击流 流数据具有如下特征 : 数据快速持续到达, 潜在大小也许是无穷无尽的 数据来源众多, 格式复杂 数据量大, 但是不十分关注存储, 一旦经过处理, 要么被丢弃, 要么被归档存储 注重数据的整体价值, 不过分关注个别数据 数据顺序颠倒, 或者不完整, 系统无法控制将要处理的新到达的数据元素的顺序

8.1.2 批量计算和实时计算 对静态数据和流数据的处理, 对应着两种截然不同的计算模式 : 批量计算和实时计算 批量计算 : 充裕时间处理静态数据, 如 Hadoop 流数据不适合采用批量计算, 因为流数据不适合用传统的关系模型建模 流数据必须采用实时计算, 响应时间为秒级 数据量少时, 不是问题, 但是, 在大数据时代, 数据格式复杂 来源众多 数据量巨大, 对实时计算提出了很大的挑战 因此, 针对流数据的实时计算 流计算, 应运而生 图 8-2 数据的两种处理模型

8.1.3 流计算概念 流计算 : 实时获取来自不同数据源的海量数据, 经过实时分析处理, 获得有价值的信息 数据采集实时分析处理结果反馈 图 8-3 流计算示意图

8.1.3 流计算概念 流计算秉承一个基本理念, 即数据的价值随着时间的流逝而降低, 如用户点击流 因此, 当事件出现时就应该立即进行处理, 而不是缓存起来进行批量处理 为了及时处理流数据, 就需要一个低延迟 可扩展 高可靠的处理引擎 对于一个流计算系统来说, 它应达到如下需求 : 高性能 : 处理大数据的基本要求, 如每秒处理几十万条数据 海量式 : 支持 TB 级甚至是 PB 级的数据规模 实时性 : 保证较低的延迟时间, 达到秒级别, 甚至是毫秒级别 分布式 : 支持大数据的基本架构, 必须能够平滑扩展 易用性 : 能够快速进行开发和部署 可靠性 : 能可靠地处理流数据

8.1.4 流计算与 Hadoop Hadoop 设计的初衷是面向大规模数据的批量处理, 每台机器并行运行 MapReduce 任务, 最后对结果进行汇总输出 MapReduce 是专门面向静态数据的批量处理的, 内部各种实现机制都为批处理做了高度优化, 不适合用于处理持续到达的动态数据 可能会想到一种 变通 的方案来降低批处理的时间延迟 将基于 MapReduce 的批量处理转为小批量处理, 将输入数据切成小的片段, 每隔一个周期就启动一次 MapReduce 作业 但这种方式也无法有效处理流数据 切分成小片段, 可以降低延迟, 但是也增加了附加开销, 还要处理片段之间依赖关系 需要改造 MapReduce 以支持流式处理 结论 : 鱼和熊掌不可兼得,Hadoop 擅长批处理, 不适合流计算

8.1.5 流计算框架 当前业界诞生了许多专门的流数据实时计算系统来满足各自需求 目前有三类常见的流计算框架和平台 : 商业级的流计算平台 开源流计算框架 公司为支持自身业务开发的流计算框架 商业级 :IBM InfoSphere Streams 和 IBM StreamBase 较为常见的是开源流计算框架, 代表如下 : Twitter Storm: 免费 开源的分布式实时计算系统, 可简单 高效 可靠地处理大量的流数据 Yahoo! S4(Simple Scalable Streaming System): 开源流计算平台, 是通用的 分布式的 可扩展的 分区容错的 可插拔的流式系统 公司为支持自身业务开发的流计算框架 : Facebook Puma Dstream( 百度 ) 银河流数据处理平台 ( 淘宝 )

8.2 流计算处理流程 8.2.1 概述 8.2.2 数据实时采集 8.2.3 数据实时计算 8.2.4 实时查询服务

8.2.1 数据处理流程 传统的数据处理流程, 需要先采集数据并存储在关系数据库等数据管理系统中, 之后由用户通过查询操作和数据管理系统进行交互 用户查询 数据管理系统 查询结果 传统的数据处理流程示意图 传统的数据处理流程隐含了两个前提 : 存储的数据是旧的 存储的静态数据是过去某一时刻的快照, 这些数据在查询时可能已不具备时效性了 需要用户主动发出查询来获取结果

8.2.1 数据处理流程 流计算的处理流程一般包含三个阶段 : 数据实时采集 数据实时计算 实时查询服务 数据实时采集 数据实时计算 用户查询 实时查询服务 查询结果 流计算处理流程示意图

8.2.2 数据实时采集 数据实时采集阶段通常采集多个数据源的海量数据, 需要保证实时性 低延迟与稳定可靠 以日志数据为例, 由于分布式集群的广泛应用, 数据分散存储在不同的机器上, 因此需要实时汇总来自不同机器上的日志数据 目前有许多互联网公司发布的开源分布式日志采集系统均可满足每秒数百 MB 的数据采集和传输需求, 如 : Facebook 的 Scribe LinkedIn 的 Kafka 淘宝的 Time Tunnel 基于 Hadoop 的 Chukwa 和 Flume 数据实时采集 数据实时计算 用户查询 实时查询服务 查询结果

8.2.2 数据实时采集 数据采集系统的基本架构一般有以下三个部分 : Agent: 主动采集数据, 并把数据推送到 Collector 部分 Collector: 接收多个 Agent 的数据, 并实现有序 可靠 高性能的转发 Store: 存储 Collector 转发过来的数据 ( 对于流计算不存储数据 ) Agent Collector Store Agent Collector Store Agent Collector Store 数据采集系统基本架构

8.2.3 数据实时计算 数据实时计算阶段对采集的数据进行实时的分析和计算, 并反馈实时结果 经流处理系统处理后的数据, 可视情况进行存储, 以便之后再进行分析计算 在时效性要求较高的场景中, 处理之后的数据也可以直接丢弃 数据流入 流处理系统实时计算 数据流出 计算结果 数据实时计算流程 数据实时采集 数据实时计算 用户查询 实时查询服务 查询结果

8.2.3 实时查询服务 实时查询服务 : 经由流计算框架得出的结果可供用户进行实时查询 展示或储存 传统的数据处理流程, 用户需要主动发出查询才能获得想要的结果 而在流处理流程中, 实时查询服务可以不断更新结果, 并将用户所需的结果实时推送给用户 虽然通过对传统的数据处理系统进行定时查询, 也可以实现不断地更新结果和结果推送, 但通过这样的方式获取的结果, 仍然是根据过去某一时刻的数据得到的结果, 与实时结果有着本质的区别 数据实时采集 数据实时计算 用户查询 实时查询服务 查询结果

8.2.3 实时查询服务 可见, 流处理系统与传统的数据处理系统有如下不同 : 流处理系统处理的是实时的数据, 而传统的数据处理系统处理的是预先存储好的静态数据 用户通过流处理系统获取的是实时结果, 而通过传统的数据处理系统, 获取的是过去某一时刻的结果 流处理系统无需用户主动发出查询, 实时查询服务可以主动将实时结果推送给用户

8.3 流计算的应用 流计算是针对流数据的实时计算, 可以应用在多种场景中, 如 Web 服务 机器翻译 广告投放 自然语言处理 气候模拟预测等 如百度 淘宝等大型网站中, 每天都会产生大量流数据, 包括用户的搜索内容 用户的浏览记录等数据 采用流计算进行实时数据分析, 可以了解每个时刻的流量变化情况, 甚至可以分析用户的实时浏览轨迹, 从而进行实时个性化内容推荐 但是, 并不是每个应用场景都需要用到流计算的 流计算适合于需要处理持续到达的流数据 对数据处理有较高实时性要求的场景

8.3.1 应用场景 1: 实时分析 传统的业务分析一般采用分布式离线计算的方式, 即将数据全部保存起来, 然后每隔一定的时间进行离线分析来得到结果 但这样会导致一定的延时, 难以保证结果的实时性 随着分析业务对实时性要求的提升, 离线分析模式已经不适合用于流数据的分析, 也不适用于要求实时响应的互联网应用场景 如淘宝网 双十一 双十二 的促销活动, 商家需要根据广告效果来即时调整广告, 这就需要对广告的受访情况进行分析 但以往采用分布式离线分析, 需要几小时甚至一天的延时才能得到分析结果 而促销活动只持续一天, 因此, 隔天才能得到的分析结果便失去了价值 虽然分布式离线分析带来的小时级的分析延时可以满足大部分商家的需求, 但随着实时性要求越来越高, 如何实现秒级别的实时分析响应成为业务分析的一大挑战

8.3.1 应用场景 1: 实时分析 针对流数据, 量子恒道 开发了海量数据实时流计算框架 Super Mario 通过该框架, 量子恒道可处理每天 TB 级的实时流数据, 并且从用户发出请求到数据展示, 整个延时控制在 2-3 秒内, 达到了实时性的要求 Log Log Log Log Time Tunnel: 日记采集系统 Super Mario: 流处理系统 HBase: 存储分析结果 Super Mario 处理流程

8.3.1 应用场景 2: 实时交通 流计算不仅为互联网带来改变, 也能改变我们的生活 如提供导航路线, 一般的导航路线并没有考虑实时的交通状况, 即便在计算路线时有考虑交通状况, 往往也只是使用了以往的交通状况数据 要达到根据实时交通状态进行导航的效果, 就需要获取海量的实时交通数据并进行实时分析 借助于流计算的实时特性, 不仅可以根据交通情况制定路线, 而且在行驶过程中, 也可以根据交通情况的变化实时更新路线, 始终为用户提供最佳的行驶路线

8.4 开源流计算框架 Storm 8.4.1 Storm 简介 8.4.2 Storm 的特点 8.4.3 Storm 设计思想 8.4.4 Storm 框架设计

8.4 开源流计算框架 Storm 以前只有政府机构和金融机构能够通过昂贵的定制系统来满足流数据实时分析计算需求 早期对于流计算的研究多数是基于对传统数据库处理的流式化, 即实时数据库, 很少研究流计算框架 Yahoo! S4 和 Twitter Storm 的开源, 改变了这个情况 在流数据处理上比 MapReduce 更有优势 批处理系统关注吞吐率, 流处理系统关注延时 Yahoo! S4 和 Twitter Storm 改变了开发实时应用的方式 以前既要关注处理逻辑, 还要解决实时数据获取 传输 存储 现在可以快速低成本搭建起实时流处理系统

8.4.1 Storm 简介 Twitter Storm 是一个免费 开源的分布式实时计算系统,Storm 对于实时计算的意义类似于 Hadoop 对于批处理的意义,Storm 可以简单 高效 可靠地处理流数据, 并支持多种编程语言 Storm 框架可以方便地与数据库系统进行整合, 从而开发出强大的实时计算系统

8.4.1 Storm 简介 Twitter 是全球访问量最大的社交网站之一,Twitter 开发 Storm 流处理框架也是为了应对其不断增长的流数据实时处理需求 实时处理系统 Storm Cassandra 数据 查询 批处理系统 Hadoop ElephantDB Twitter 的分层数据处理架构

8.4.2 Storm 的特点 Storm 可用于许多领域中, 如实时分析 在线机器学习 持续计算 远程 RPC 数据提取加载转换等 Storm 具有以下主要特点 : 整合性 :Storm 可方便地与队列系统和数据库系统进行整合 简易的 API:Storm 的 API 在使用上即简单又方便 可扩展性 :Storm 的并行特性使其可以运行在分布式集群中 容错性 :Storm 可自动进行故障节点的重启 任务的重新分配 可靠的消息处理 :Storm 保证每个消息都能完整处理 支持各种编程语言 :Storm 支持使用各种编程语言来定义任务 快速部署 :Storm 可以快速进行部署和使用 免费 开源 :Storm 是一款开源框架, 可以免费使用

8.4.3 Storm 设计思想 Storm 主要术语包括 Streams Spouts Bolts Topology 和 Stream Groupings Streams:Storm 将流数据 Stream 描述成一个无限的 Tuple 序列, 这些 Tuple 序列会以分布式的方式并行地创建和处理 Streams 无界的 Tuple 序列 Tuple Tuple Tuple Tuple Tuple 每个 tuple 是一堆值, 每个值有一个名字, 并且每个值可以是任何类型 Tuple 本来应该是一个 Key-Value 的 Map, 由于各个组件间传递的 tuple 的字段名称已经事先定义好了, 所以 Tuple 只需要按序填入各个 Value, 所以就是一个 Value List( 值列表 ) Field1 Field2 Field3 Field4

8.4.3 Storm 设计思想 Spout:Storm 认为每个 Stream 都有一个源头, 并把这个源头抽象为 Spout 通常 Spout 会从外部数据源 ( 队列 数据库等 ) 读取数据, 然后封装成 Tuple 形式, 发送到 Stream 中 Spout 是一个主动的角色, 在接口内部有个 nexttuple 函数,Storm 框架会不停的调用该函数 Spouts Streams 的来源 Tuple Tuple Tuple Tuple Tuple

8.4.3 Storm 设计思想 Bolt:Storm 将 Streams 的状态转换过程抽象为 Bolt Bolt 即可以处理 Tuple, 也可以将处理后的 Tuple 作为新的 Streams 发送给其他 Bolt Bolt 可以执行过滤 函数操作 Join 操作数据库等任何操作 Bolt 是一个被动的角色, 其接口中有一个 execute(tuple input) 方法, 在接收到消息之后会调用此函数, 用户可以在此方法中执行自己的处理逻辑 Tuple Tuple Tuple Tuple Tuple Bolts 处理 Tuples 创建新 Streams Tuple Tuple Tuple Tuple Tuple Tuple Tuple Tuple Tuple Tuple

8.4.3 Storm 设计思想 Topology:Storm 将 Spouts 和 Bolts 组成的网络抽象成 Topology, 它可以被提交到 Storm 集群执行 Topology 可视为流转换图, 图中节点是一个 Spout 或 Bolt, 边则表示 Bolt 订阅了哪个 Stream 当 Spout 或者 Bolt 发送元组时, 它会把元组发送到每个订阅了该 Stream 的 Bolt 上进行处理 Topology 里面的每个处理组件 (Spout 或 Bolt) 都包含处理逻辑, 而组件之间的连接则表示数据流动的方向 Topology 里面的每一个组件都是并行运行的 在 Topology 里面可以指定每个组件的并行度, Storm 会在集群里面分配那么多的线程来同时计算 在 Topology 的具体实现上,Storm 中的 Topology 定义仅仅是一些 Thrift 结构体 ( 二进制高性能的通信中间件 ), 支持各种编程语言进行定义 Spout Spout Topology Bolt Bolt Bolt Bolt Bolt

8.4.3 Storm 设计思想 Stream Groupings:Storm 中的 Stream Groupings 用于告知 Topology 如何在两个组件间 ( 如 Spout 和 Bolt 之间, 或者不同的 Bolt 之间 ) 进行 Tuple 的传送 每一个 Spout 和 Bolt 都可以有多个分布式任务, 一个任务在什么时候 以什么方式发送 Tuple 就是由 Stream Groupings 来决定的 Spout Stream Groupings Bolt A Bolt B Bolt C

8.4.3 Storm 设计思想 目前,Storm 中的 Stream Groupings 有如下几种方式 : (1)ShuffleGrouping: 随机分组, 随机分发 Stream 中的 Tuple, 保证每个 Bolt 的 Task 接收 Tuple 数量大致一致 (2)FieldsGrouping: 按照字段分组, 保证相同字段的 Tuple 分配到同一个 Task 中 (3)AllGrouping: 广播发送, 每一个 Task 都会收到所有的 Tuple (4)GlobalGrouping: 全局分组, 所有的 Tuple 都发送到同一个 Task 中 (5)NonGrouping: 不分组, 和 ShuffleGrouping 类似, 当前 Task 的执行会和它的被订阅者在同一个线程中执行 (6)DirectGrouping: 直接分组, 直接指定由某个 Task 来执行 Tuple 的处理

8.4.4 Storm 框架设计 Storm 运行任务的方式与 Hadoop 类似 :Hadoop 运行的是 MapReduce 作业, 而 Storm 运行的是 Topology 但两者的任务大不相同, 主要的不同是 :MapReduce 作业最终会完成计算并结束运行, 而 Topology 将持续处理消息 ( 直到人为终止 ) Storm 和 Hadoop 架构组件功能对应关系 Hadoop Storm 应用名称 Job Topology 系统角色 JobTracker Nimbus TaskTracker Supervisor 组件接口 Map/Reduce Spout/Bolt

8.4.4 Storm 框架设计 Storm 集群采用 Master Worker 的节点方式 : Master 节点运行名为 Nimbus 的后台程序 ( 类似 Hadoop 中的 JobTracker ), 负责在集群范围内分发代码 为 Worker 分配任务和监测故障 Worker 节点运行名为 Supervisor 的后台程序, 负责监听分配给它所在机器的工作, 即根据 Nimbus 分配的任务来决定启动或停止 Worker 进程, 一个 Worker 节点上同时运行若干个 Worker 进程

8.4.4 Storm 框架设计 Storm 使用 Zookeeper 来作为分布式协调组件, 负责 Nimbus 和多个 Supervisor 之间的所有协调工作 借助于 Zookeeper, 若 Nimbus 进程或 Supervisor 进程意外终止, 重启时也能读取 恢复之前的状态并继续工作, 使得 Storm 极其稳定 Supervisor Zookeeper Supervisor Worker 进程 Nimbus Zookeeper Supervisor Worker 进程 Zookeeper Supervisor Worker 进程 Supervisor Storm 集群架构示意图

8.4.4 Storm 框架设计 (1)worker: 每个 worker 进程都属于一个特定的 Topology, 每个 Supervisor 节点的 worker 可以有多个, 每个 worker 对 Topology 中的每个组件 (Spout 或 Bolt) 运行一个或者多个 executor 线程来提供 task 的运行服务 (2)executor:executor 是产生于 worker 进程内部的线程, 会执行同一个组件的一个或者多个 task (3)task: 实际的数据处理由 task 完成, 在 Topology 的生命周期中, 每个组件的 task 数目是不会发生变化的, 而 executor 的数目却不一定 executor 数目小于等于 task 的数目, 默认情况下, 二者是相等的 Worker 进程 每台 Supervisor 上运行着若干 worker 进程 Task Task Executor 线程 Task Task 每个 Worker 进程中运行着若干个 Executor 线程 Worker Executor 和 Task 的关系 每个 Executor 线程里面运行着若干个相同的 Task(Spout/Bolt)

8.4.4 Storm 框架设计 基于这样的架构设计,Storm 的工作流程如下图所示 : 所有 Topology 任务的提交必须在 Storm 客户端节点上进行, 提交后, 由 Nimbus 节点分配给其他 Supervisor 节点进行处理 Nimbus 节点首先将提交的 Topology 进行分片, 分成一个个 Task, 分配给相应的 Supervisor, 并将 Task 和 Supervisor 相关的信息提交到 Zookeeper 集群上 Supervisor 会去 Zookeeper 集群上认领自己的 Task, 通知自己的 Worker 进程进行 Task 的处理 说明 : 在提交了一个 Topology 之后,Storm 就会创建 Spout/Bolt 实例并进行序列化 之后, 将序列化的组件发送给所有的任务所在的机器 ( 即 Supervisor 节点 ), 在每一个任务上反序列化组件 1. 提交 Topology 2. 将任务存储在 Zookeeper 中 Client Task Nimbus Supervisor Worker Task Storm 工作流程示意图 Zookeeper 3. 获取分配的任务, 并启动 Worker 4. Worker 进程执行具体的任务

8.5 Spark Streaming 8.5.1 Spark Streaming 设计 8.5.2 Spark Streaming 与 Storm 的对比

8.5.1 Spark Streaming 设计 Spark Streaming 可整合多种输入数据源, 如 Kafka Flume HDFS, 甚至是普通的 TCP 套接字 经处理后的数据可存储至文件系统 数据库, 或显示在仪表盘里 Kafka Flume HDFS TCP socket Spark Streaming HDFS Databases Dashboards 图 13 Spark Streaming 支持的输入 输出数据源

8.5.1 Spark Streaming 设计 Spark Streaming 的基本原理是将实时输入数据流以时间片 ( 秒级 ) 为单位进行拆分, 然后经 Spark 引擎以类似批处理的方式处理每个时间片数据 input data stream Spark Streaming batches of input data Spark Engine batches of processed data 图 14 Spark Streaming 执行流程

8.5.1 Spark Streaming 设计 Spark Streaming 最主要的抽象是 DStream(Discretized Stream, 离散化数据流 ), 表示连续不断的数据流 在内部实现上,Spark Streaming 的输入数据按照时间片 ( 如 1 秒 ) 分成一段一段的 DStream, 每一段数据转换为 Spark 中的 RDD, 并且对 DStream 的操作都最终转变为对相应的 RDD 的操作 RDD @ time 1 RDD @ time 2 RDD @ time 3 RDD @ time 4 Lines DStream lines from time 0 to 1 lines from time 1 to 2 lines from time 2 to 3 lines from time 3 to 4 flatmap operation words DStream words from time 0 to 1 words from time 1 to 2 words from time 2 to 3 words from time 3 to 4 RDD @ result 1 RDD @ result 2 RDD @ result 3 RDD @ result 4 图 15 DStream 操作示意图

8.5.2 Spark Streaming 与 Storm 的对比 Spark Streaming 和 Storm 最大的区别在于,Spark Streaming 无法实现毫秒级的流计算, 而 Storm 可以实现毫秒级响应 Spark Streaming 构建在 Spark 上, 一方面是因为 Spark 的低延迟执行引擎 (100ms+) 可以用于实时计算, 另一方面, 相比于 Storm,RDD 数据集更容易做高效的容错处理 Spark Streaming 采用的小批量处理的方式使得它可以同时兼容批量和实时数据处理的逻辑和算法, 因此, 方便了一些需要历史数据和实时数据联合分析的特定应用场合

8.6 Samza 8.6.1 基本概念 8.6.2 系统架构

8.6.1 基本概念 1. 作业一个作业 (Job) 是对一组输入流进行处理转化成输出流的程序 Input Streams Samza Job output Streams

8.6.1 基本概念 2. 分区 Samza 的流数据单位既不是 Storm 中的元组, 也不是 Spark Streaming 中的 DStream, 而是一条条消息 Samza 中的每个流都被分割成一个或多个分区, 对于流里的每一个分区而言, 都是一个有序的消息序列, 后续到达的消息会根据一定规则被追加到其中一个分区里 A Partitioned Stream Partition 0 0 1 2 3 4 5 6 7 8 9 Partition 1 next append 0 1 2 3 4 5 6 7 Partition 2 0 1 2 3 4 5 6 7 8

8.6.1 基本概念 3. 任务 一个作业会被进一步分割成多个任务 ( Task) 来执行, 其中, 每个任务负责处理作业中的一个分区 分区之间没有定义顺序, 从而允许每一个任务独立执行 YARN 调度器负责把任务分发给各个机器, 最终, 一个工作中的多个任务会被分发到多个机器进行分布式并行处理 Input Stream A 0 Samza Job 1 Input Stream B Task 1 Task 2 0 0 1 0 1 1 Output Stream C

8.6.1 基本概念 4. 数据流图 一个数据流图是由多个作业构成的, 其中, 图中的每个节点表示包含数据的流, 每条边表示数据传输 多个作业串联起来就完成了流式的数据处理流程 由于采用了异步的消息订阅分发机制, 不同任务之间可以独立运行 Stream A Stream B Stream C Job 1 Job 2 Stream D Stream E Job 3 Stream F

8.6.2 系统架构 Samza 系统架构主要包括 流数据层 (Kafka) 执行层 (YARN) 处理层 (Samza API) 流处理层和执行层都被设计成可插拔的, 开发人员可以使用其他框架来替代 YARN 和 Kafka 表 MapReduce 批处理架构和 Samza 流处理架构的类比 MapReduce 批处理架构 Samza 流处理架构 数据层 HDFS Kafka 执行层 YARN YARN 处理层 MapReduce Samza API

8.6.2 系统架构 处理分析过程如下 : Samza 客户端需要执行一个 Samza 作业时, 它会向 YARN 的 ResouceManager 提交作业请求 ResouceManager 通过与 NodeManager 沟通为该作业分配容器 ( 包含了 CPU 内存等资源 ) 来运行 Samza ApplicationMaster Samza ApplicationMaster 进一步向 ResourceManager 申请运行任务的容器 获得容器后,Samza ApplicationMaster 与容器所在的 NodeManager 沟通, 启动该容器, 并在其中运行 Samza Task Runner Samza Task Runner 负责执行具体的 Samza 任务, 完成流数据处理分析 Samza YARN Client NM Samza AM Kafka Broker RM NM Samza Task Runner Kafka Broker

8.7 Storm Spark Streaming 和 Samza 的应用场景 从编程的灵活性来讲,Storm 是比较理想的选择, 它使用 Apache Thrift, 可以用任何编程语言来编写拓扑结构 (Topology) 当需要在一个集群中把流计算和图计算 机器学习 SQL 查询分析等进行结合时, 可以选择 Spark Streaming, 因为, 在 Spark 上可以统一部署 Spark SQL,Spark Streaming MLlib,GraphX 等组件, 提供便捷的一体化编程模型 当有大量的状态需要处理时, 比如每个分区都有数十亿个元组, 则可以选择 Samza 当应用场景需要毫秒级响应时, 可以选择 Storm 和 Samza, 因为 Spark Streaming 无法实现毫秒级的流计算

8.8 Storm 编程实践 8.8.1 编写 Storm 程序 8.8.2 安装 Storm 的基本过程 8.8.3 运行 Storm 程序 Storm 上机实践详细过程, 请参考厦门大学数据库实验室建设的 中国高校大数据课程公共服务平台 中的 大数据课程学生服务站 中的 学习指南 栏目 : 学生服务站地址 :http://dblab.xmu.edu.cn/post/4331/ 学习指南栏目中包含了 第八章流计算学习指南 http://dblab.xmu.edu.cn/blog/767/ 扫一扫访问学生服务站

8.8.1 编写 Storm 程序 程序任务 : 单词统计 基于 Storm 的单词统计在形式上与基于 MapReduce 的单词统计是类似的,MapReduce 使用的是 Map 和 Reduce 的抽象, 而 Storm 使用的是 Spout 和 Bolt 的抽象 Storm 进行单词统计的整个流程 : 从 Spout 中发送 Stream( 每个英文句子为一个 Tuple) 用于分割单词的 Bolt 将接收的句子分解为独立的单词, 将单词作为 Tuple 的字段名发送出去 用于计数的 Bolt 接收表示单词的 Tuple, 并对其进行统计 输出每个单词以及单词出现过的次数

8.8.1 编写 Storm 程序 ["the"] ["the", 1] ["cow"] ["cow", 1] ["the cow jumped over the moon"] ["jumped"] ["over"] ["jumped", 1] ["over", 1] ["the"] ["the", 2] ["moon"] ["moon", 1] 一个句子经 Storm 的单词统计得出的结果

8.8.1 编写 Storm 程序 Storm 的编程模型非常简单, 如下代码即定义了整个单词统计 Topology 的整体逻辑 import org.apache.storm.config; Import public class WordCountTopology { public static class RandomSentenceSpout extends BaseRichSpout { } public static class SplitSentence extends ShellBolt implements IRichBolt { } public static class WordCount extends BaseBasicBolt { } } } public static void main(string[] args) throws Exception { TopologyBuilder builder = new TopologyBuilder(); builder.setspout("sentences", new RandomSentenceSpout(), 5); builder.setbolt("split", new SplitSentence(), 8).shuffleGrouping("sentences"); builder.setbolt("count", new WordCount(), 12).fieldsGrouping("split", new Fields("word"));

8.8.1 编写 Storm 程序 main() 函数中的处理逻辑 Topology 中仅定义了整体的计算逻辑, 还需要定义具体的处理函数 具体的处理函数可以使用任一编程语言来定义, 甚至也可以结合多种编程语言来实现

8.8.1 编写 Storm 程序 RandomSentenceSpout 类 备注 : 为简单起见,RandomSentenceSpout 省略了类中的一些方法 public class RandomSentenceSpout extends BaseRichSpout { SpoutOutputCollector _collector; Random _rand; @Override public void nexttuple() { } Utils.sleep(100); String[] sentences = new String[]{ "the cow jumped over the moon", "an apple a day keeps the doctor away", "four score and seven years ago", "snow white and the seven dwarfs", "i am at two with nature" }; String sentence = sentences[_rand.nextint(sentences.length)]; _collector.emit(new Values(sentence)); } @Override public void declareoutputfields(outputfieldsdeclarer declarer) { } declarer.declare(new Fields( sentences"));

8.8.1 编写 Storm 程序 SplitSentence 类 如 SplitSentence() 方法虽然是通过 Java 语言定义的, 但具体的操作可通过 Python 脚本来完成 Topology 里面的每个组件必须定义它要发射的 Tuple 的每个字段

8.8.1 编写 Storm 程序 splitsentence.py Python 脚本 splitsentence.py 定义了一个简单的单词分割方法, 即通过空格来分割单词 分割后的单词通过 emit() 方法以 Tuple 的形式发送给订阅了该 Stream 的 Bolt 进行接收和处理

8.8.1 编写 Storm 程序 WordCount 类 单词统计的具体逻辑 : 首先判断单词是否统计过, 若未统计过, 需先将 count 值置为 0 若单词已统计过, 则每出现一次该单词,count 值就加 1

8.8.1 编写 Storm 程序 上述虽然是一个简单的单词统计, 但对其进行扩展, 便可应用到许多场景中, 如微博中的实时热门话题 Twitter 也正是使用了 Storm 框架实现了实时热门话题 CountsBolt Tweets (topic) (topic, count) IntermediateRankingsBolt TwitterStreamingTopicSpout TotalRankingsBolt (rankings) RankingReportBolt (rankings) (rankings) Websites Twitter 实时热门话题处理流程示意图

8.8.2 安装 Storm 的基本过程 本实例中 Storm 具体运行环境如下 : CentOS 6.4 Storm 0.9.6 Java JDK 1.7 ZooKeeper 3.4.6 Python 2.6 备注 :CentOS 中已默认安装了 Python 2.6, 我们还需要安装 JDK 环境以及分布式应用程序协调服务 Zookeeper 安装 Storm 的基本过程如下 : 第一步 : 安装 Java 环境 第二步 : 安装 Zookeeper 第三步 : 安装 Storm( 单机 ) 第四步 : 关闭 Storm Storm 上机实践详细过程, 请参考厦门大学数据库实验室建设的 中国高校大数据课程公共服务平台 中的 大数据课程学生服务站 中的 学习指南 栏目 : 学生服务站地址 : http://dblab.xmu.edu.cn/post/4331/ 学习指南栏目中包含了 第八章流计算学习指南 http://dblab.xmu.edu.cn/blog/767/

8.8.2 安装 Storm 的基本过程 第一步 : 安装 Java 环境 Storm 运行需要 Java 环境, 可选择 Oracle 的 JDK, 或是 OpenJDK, 现在一般 Linux 系统默认安装的基本是 OpenJDK, 如 CentOS 6.4 就默认安装了 OpenJDK 1.7 但需要注意的是,CentOS 6.4 中默认安装的只是 Java JRE, 而不是 JDK, 为了开发方便, 我们还是需要通过 yum 进行安装 JDK $ sudo yum install java-1.7.0-openjdk java-1.7.0-openjdk-devel 接着需要配置一下 JAVA_HOME 环境变量, 为方便, 可以在 ~/.bashrc 中进行设置

8.8.2 安装 Storm 的基本过程 第二步 : 安装 Zookeeper 到官网下载 Zookeeper, 比如下载 zookeeper-3.4.6.tar.gz 下载后执行如下命令进行安装 zookeeper( 将命令中 3.4.6 改为你下载的版本 ): $ sudo tar -zxf ~/ 下载 /zookeeper-3.4.6.tar.gz -C /usr/local $ cd /usr/local $ sudo mv zookeeper-* zookeeper # 修改目录名称方便使用 $ sudo chown -R hadoop:hadoop./zookeeper # 此处的 hadoop 为你的用户名 chown 命令让 hadoop 用户拥有 zookeeper 目录下的所有文件的权限

8.8.2 安装 Storm 的基本过程 第二步 : 安装 Zookeeper( 续 ) 接着执行如下命令进行 zookeeper 配置 : $ cd /usr/local/zookeeper $ mkdir tmp $ cp./conf/zoo_sample.cfg./conf/zoo.cfg $ vim./conf/zoo.cfg 进入 zoo.cfg 文件编辑状态后, 将当中的 datadir=/tmp/zookeeper 更改为 datadir=/usr/local/zookeeper/tmp 接着执行 : $./bin/zkserver.sh start

8.8.2 安装 Storm 的基本过程 第三步 : 安装 Storm( 单机 ) 到官网下载 Storm, 比如 Storm0.9.6 下载后执行如下命令进行安装 Storm: $ sudo tar -zxf ~/ 下载 /apache-storm-0.9.6.tar.gz -C /usr/local $ cd /usr/local $ sudo mv apache-storm-0.9.6 storm $ sudo chown -R hadoop:hadoop./storm # 此处的 hadoop 为你的用户名 接着执行如下命令进行 Storm 配置 : $ cd /usr/local/storm $ vim./conf/storm.yaml 备注 :storm 的运行有两种模式 : 本地模式和分布式模式. 在本地模式中, storm 用一个进程里面的线程来模拟所有的 spout 和 bolt. 本地模式对开发和测试来说比较有用 在分布式模式下, storm 由一堆机器组成 当提交 topology 给 master 的时候, master 负责分发代码并且负责给 topolgoy 分配工作进程 如果一个工作进程挂掉了, master 节点会把它重新分配到其它节点

8.8.2 安装 Storm 的基本过程 第三步 : 安装 Storm( 单机 ) ( 续 ) 修改其中的 storm.zookeeper.servers 和 nimbus.host 两个配置项, 即取消掉注释且都修改值为 127.0.0.1( 我们只需要在单机上运行 ), 如下图所示 然后就可以启动 Storm 了 执行如下命令启动 nimbus 后台进程 : $./bin/storm nimbus

8.8.2 安装 Storm 的基本过程 第三步 : 安装 Storm( 单机 ) ( 续 ) 启动 nimbus 后, 终端被该进程占用了, 不能再继续执行其他命令了 因此我们需要另外开启一个终端, 然后执行如下命令启动 supervisor 后台进程 : $ # 需要另外开启一个终端 $ /usr/local/storm/bin/storm supervisor 同样的, 启动 supervisor 后, 我们还需要开启另外的终端才能执行其他命令 另外, 我们可以使用 jps 命令检查是否成功启动, 若成功启动会显示 nimbus supervisor QuorumPeeMain (QuorumPeeMain 是 zookeeper 的后台进程, 若显示 config_value 表明 nimbus 或 supervisor 还在启动中 ), 如下图所示

8.8.2 安装 Storm 的基本过程 第四步 : 关闭 Storm 之前启动的 nimbus 和 supervisor 占用了两个终端窗口, 切换到这两个终端窗口, 按键盘的 Ctrl+C 可以终止进程, 终止后, 也就相当于关闭了 Storm

8.8.3 运行 Storm 实例 运行 Storm 计算任务, 就是提交 Topology 运行一个 Topology 是很简单的 首先, 把所有的代码以及所依赖的 jar 打进一个 jar 包 然后运行类似下面的这个命令 strom jar all-your-code.jar backtype.storm.mytopology arg1 arg2 storm jar 负责连接到 nimbus 并且上传 jar 文件

8.8.3 运行 Storm 实例 Storm 中自带了一些例子, 我们可以执行一下 WordCount 例子来感受一下 Storm 的执行流程 执行如下命令 : $ /usr/local/storm/bin/storm jar /usr/local/storm/examples/stormstarter/storm-starter-topologies-0.9.6.jar storm.starter.wordcounttopology 该程序是不断地取如下四句英文句子中的一句作为数据源, 然后发送给 bolt 来统计单词出现的次数 { "the cow jumped over the moon", "an apple a day keeps the doctor away", "four score and seven years ago", "snow white and the seven dwarfs", "i am at two with nature" }

本章小结 本章首先介绍了流计算的基本概念和需求 流数据即持续到达的大量数据, 对流数据的处理强调实时性, 一般要求为秒级 MapReduce 框架虽然广泛应用于大数据处理中, 但其面向的是海量数据的离线处理, 并不适合用于处理持续到达的流数据 本章阐述了流计算的处理流程, 一般包括数据实时采集 数据实时计算和实时查询服务三个部分, 并比较其与传统的数据处理流程的不同 流计算处理的是实时数据, 而传统的批处理则处理的是预先存储好的静态数据 流计算可应用在多个场景中, 如实时业务分析, 流计算带来的实时性特点, 可以大大增加实时数据的价值, 为业务分析带来质的提升 本章接着介绍了流计算框架 Storm 的设计思想和架构设计 Storm 流处理框架具有可扩展性 高容错性 能可靠地处理消息的特点, 使用简单, 学习和开发成本较低 Storm 框架对设计概念进行了抽象化, 其主要术语包括 Streams Spouts Bolts Topology 和 Stream Groupings, 在 Topology 中定义整体任务的处理逻辑, 再通过 Bolt 具体执行,Stream Groupings 则定义了 Tuple 如何在不同组件间进行传输, 通过一个单词统计的实例来加深对 Storm 框架的了解 介绍了开源流计算框架 Spark Streaming 和 Samza, 并和 Storm 做了对比 介绍了 Storm 安装和运行程序

附录 : 主讲教师 主讲教师 : 林子雨 单位 : 厦门大学计算机科学系 E-mail: ziyulin@xmu.edu.cn 个人网页 :http://www.cs.xmu.edu.cn/linziyu 数据库实验室网站 :http://dblab.xmu.edu.cn 扫一扫访问个人主页 林子雨, 男,1978 年出生, 博士 ( 毕业于北京大学 ), 现为厦门大学计算机科学系助理教授 ( 讲师 ), 曾任厦门大学信息科学与技术学院院长助理 晋江市发展和改革局副局长 中国高校首个 数字教师 提出者和建设者, 厦门大学数据库实验室负责人, 厦门大学云计算与大数据研究中心主要建设者和骨干成员,2013 年度厦门大学奖教金获得者 主要研究方向为数据库 数据仓库 数据挖掘 大数据 云计算和物联网, 并以第一作者身份在 软件学报 计算机学报 和 计算机研究与发展 等国家重点期刊以及国际学术会议上发表多篇学术论文 作为项目负责人主持的科研项目包括 1 项国家自然科学青年基金项目 (No.61303004) 1 项福建省自然科学青年基金项目 (No.2013J05099) 和 1 项中央高校基本科研业务费项目 (No.2011121049), 同时, 作为课题负责人完成了国家发改委城市信息化重大课题 国家物联网重大应用示范工程区域试点泉州市工作方案 2015 泉州市互联网经济调研等课题 编著出版中国高校第一本系统介绍大数据知识的专业教材 大数据技术原理与应用 并成为畅销书籍, 编著并免费网络发布 40 余万字中国高校第一本闪存数据库研究专著 闪存数据库概念与技术 ; 主讲厦门大学计算机系本科生课程 数据库系统原理 和研究生课程 分布式数据库 大数据技术基础 具有丰富的政府和企业信息化培训经验, 曾先后给中国移动通信集团公司 福州马尾区政府 福建省物联网科学研究院 石狮市物流协会 厦门市物流协会 福建龙岩卷烟厂等多家单位和企业开展信息化培训, 累计培训人数达 2000 人以上

附录 : 大数据学习教材推荐 大数据技术原理与应用 概念 存储 处理 分析与应用, 由厦门大学计算机科学系林子雨博士编著, 是中国高校第一本系统介绍大数据知识的专业教材 扫一扫访问教材官网 全书共有 13 章, 系统地论述了大数据的基本概念 大数据处理架构 Hadoop 分布式文件系统 HDFS 分布式数据库 HBase NoSQL 数据库 云数据库 分布式并行编程模型 MapReduce 流计算 图计算 数据可视化以及大数据在互联网 生物医学和物流等各个领域的应用 在 Hadoop HDFS HBase 和 MapReduce 等重要章节, 安排了入门级的实践操作, 让读者更好地学习和掌握大数据关键技术 本书可以作为高等院校计算机专业 信息管理等相关专业的大数据课程教材, 也可供相关技术人员参考 学习 培训之用 欢迎访问 大数据技术原理与应用 概念 存储 处理 分析与应用 教材官方网站 : http://dblab.xmu.edu.cn/post/bigdata

附录 : 中国高校大数据课程公共服务平台 http://dblab.xmu.edu.cn/post/bigdata-teaching-platform/ 扫一扫访问平台主页 扫一扫观看 3 分钟 FLASH 动画宣传片

Department of Computer Science, Xiamen University, 2016