【作者主页】Francek Chen【专栏介绍】⌈ ⌈⌈大数据技术原理与应用⌋ ⌋⌋专栏系统介绍大数据的相关知识分为大数据基础篇、大数据存储与管理篇、大数据处理与分析篇、大数据应用篇。内容包含大数据概述、大数据处理架构Hadoop、分布式文件系统HDFS、分布式数据库HBase、NoSQL数据库、云数据库、MapReduce、Hadoop再探讨、数据仓库Hive、Spark、流计算、Flink、图计算、数据可视化以及大数据在互联网领域、生物医学领域的应用和大数据的其他应用。【GitCode】专栏资源保存在我的GitCode仓库https://gitcode.com/Morse_Chen/BigData_principle_application。文章目录一、Pig二、Tez三、Kafka小结本文介绍 Hadoop 生态系统中具有代表性的功能组件包括 Pig、Tez 和 Kafka 等。一、PigPig 是 Hadoop 生态系统的一个组件提供了类似 SQL 的 Pig Latin包含 Filter、GroupBy、Join、OrderBy 等操作同时也支持用户自定义函数允许用户通过编写简单的脚本来实现复杂的数据分析而不需要编写复杂的 MapReduce 应用程序。Pig 会自动把用户编写的脚本转换成 MapReduce 作业在 Hadoop 集群上运行而且具备对生成的 MapReduce 程序进行自动优化的功能所以用户在编写 Pig 程序的时候不需要关心程序的运行效率这就大大缩短了用户编程时间。因此通过配合使用 Pig 和 Hadoop用户在处理海量数据时就可以达到事半功倍的效果。这比使用 Java、C等语言编写 MapReduce 程序的难度要小很多并且用更少的代码量实现了相同的数据处理分析功能。Pig 可以加载数据、表达转换数据以及存储最终结果因此在企业数据分析系统实际应用中见图1Pig 通常用于数据提取、转换和加载Extraction、Transformation、LoadingETL过程即来自各个不同数据源的数据被收集过来以后使用 Pig 进行统一加工处理然后加载到数据仓库 Hive 中由 Hive 实现对海量数据的分析。需要特别指出的是每种数据分析工具都有一定的局限性Pig 的设计和 MapReduce 一样都是面向批处理的因此 Pig 并不适合所有的数据处理任务特别是当需要查询大数据集中的一小部分数据时Pig 仍然需要对整个或绝大部分数据集进行扫描因此实现性能不会很好。图1 Pig在企业数据分析系统实际应用中的作用Pig 语句通常按照如下的格式来编写。通过 LOAD 语句从文件系统读取数据。通过一系列“转换”语句对数据进行处理。通过一条 STORE 语句把处理结果输出到文件系统中或者使用 DUMP 语句把处理结果输出到屏幕上。下面是一个采用 Pig Latin 编写的应用程序实例实现对用户访问网页情况的统计分析visits load /data/visits as (user, url, time); //导入用户访问日志visits gVisits group visits by url; //根据网址url对用户访问数据进行分组 visitCounts foreach gVisits generate url, count(visits); //对于每个url计算用户访问量 //上面语句执行后得到的表的结构visitCounts(url,visits) urlInfo load /data/urlInfo as (url, category, pRank); //导入用户信息 visitCounts join visitCounts by url, urlInfo by url; //对visitCounts和urlInfo表进行连接操作 //上面语句执行后得到的连接结果表的结构visitCounts(url,visits,category,pRank) gCategories group visitCounts by category; //根据用户类别进行分组 topUrls foreach gCategories generate top(visitCounts,10); //每个用户类别取访问量TOP10 store topUrls into /data/topUrls; //把访问量排名信息写入topUrls对于上述 Pig Latin 脚本Pig 会自动转换成 MapReduce 作业如图2所示。图2中group by 和 join 操作都“跨越”了 Map 和 Reduce 两个阶段这是因为 group by 和 join 操作都涉及 Shuffle 过程Shuffle 过程包含了 Map 端和 Reduce 端所以图中表示 group by 和 join 操作的矩形框与 Map 和 Reduce 两个阶段都存在重叠区域。图2 从Pig Latin脚本转化得到的MapReduce作业当数据查询只面向相关技术人员并且属于即时性的数据处理需求时比较适合采用 Pig 编写一个脚本来完成快速运行处理从而避免创建表等相关操作。目前Pig 在很多公司得到了应用在 Yahoo!中 90%以上的 MapReduce 作业是 Pig 生成的Twitter 80%以上的 MapReduce 作业是 Pig 生成的LinkedIn 大部分 MapReduce 作业是 Pig 生成的。Pig 其他主要用户还包括 Salesforce、Nokia、AOL 和 ComScore 等。二、TezTez 是 Apache 开源的支持 DAG 作业的计算框架直接源于 MapReduce 框架。其核心思想是将 Map 和 Reduce 两个操作进一步进行拆分即将 Map 拆分成 Input、Processor、Sort、Merge 和 Output 等将 Reduce 拆分成 Input、Shuffle、Sort、Merge、Processor 和 Output 等经过分解后的这些元操作可以进行任意组合产生新的操作经过一些控制程序组装后就可形成一个大的 DAG 作业。通过 DAG 作业的方式运行 MapReduce 作业提供了程序运行的整体处理逻辑就可以去除工作流当中多余的 Map 阶段减少不必要的操作提升数据处理的性能。Hortonworks 把 Tez 应用到数据仓库 Hive 的优化中使得性能提升了约 100 倍。下面以一个具体实例说明 Tez 的优化效果。假设有 3 个表 a、b 和 c表 a 的属性包括 state、id 和 itemId表 b 的属性包括 id 和其他属性表 c 的属性包括 itemId 和 price。以下是数据仓库 Hive 中执行数据分析的一段 HiveQL 语句其功能是对 3 个表在公共属性上进行连接操作并根据 state 属性进行分组统计每个分组的元组个数和平均价格SELECTa.state,COUNT(*),AVERAGE(c.price)FROMaJOINbON(a.idb.id)JOINcON(a.itemIdc.itemId)GROUPBYa.state上述 HiveQL 语句在 MapReduce 和 Tez 中的执行情况对比如图3所示。在 MapReduce 框架中执行时需要采用 3 个 MapReduce 作业才能完成每个作业完成以后都需要写入分布式文件系统 HDFS 中供下一个作业读取带来了较大的处理延迟而在 Tez 框架中执行时只需要一个 Tez 作业就可以完成处理全过程不需要把数据写入 HDFS 中可以直接在不同的 Map 和 Reduce 任务之间传输数据同时减少了一些不必要的 MapReduce 操作从而大大提升了程序执行效率。图3 HiveQL语句在MapReduce和Tez中的执行情况对比在 Hadoop 2.0 生态系统中MapReduce、Hive、Pig 等计算框架都需要最终以 MapReduce 任务的形式执行数据分析因此 Tez 框架可以发挥重要的作用。可以让 Tez 框架运行在 YARN 框架之上如图4所示然后让 MapReduce、Pig 和 Hive 等计算框架运行在 Tez 框架之上从而借助于 Tez 框架实现对 MapReduce、Pig 和 Hive 等的性能优化更好地解决现有 MapReduce 框架在迭代计算如 PageRank 计算和交互式计算方面存在的问题。图4 Tez框架在Hadoop生态系统中的作用可以看出Tez 在解决 Hive、Pig 延迟大、性能低等问题的思路上和那些支持实时交互式查询分析的产品如 Impala、Dremel 和 Drill 等是不同的。Impala、Dremel 和 Drill 解决问题的思路是抛弃 MapReduce 计算框架不再将类似 SQL 语句的 HiveQL 或者 Pig 语句翻译成 MapReduce 程序而是采用与商用并行关系数据库类似的分布式查询引擎直接从 HDFS 或者 HBase 中用 SQL 语句查询数据从而大大降低了延迟很好地满足了实时查询的要求。Tez 则不同比如针对 Hive 数据仓库进行优化的“TezHive”解决方案其仍采用 MapReduce 计算框架但是对 DAG 的作业依赖关系进行了裁剪并将多个小作业合并成一个大作业这样不仅计算量减少了而且读写 HDFS 次数也会大大减少。三、KafkaKafka 是由 LinkedIn 开发的一种高吞吐量的分布式发布/订阅消息系统用户通过 Kafka 系统可以发布大量的消息同时能实时订阅消费消息。Kafka 设计的初衷是构建一个可以处理海量日志、用户行为和网站运营统计等的数据处理框架。为了满足上述应用需求系统需要同时提供实时在线处理的低延迟和批量离线处理的高吞吐量。现有的一些消息队列框架通常设计了完备的机制来保证消息传输的可靠性但是由此会带来较大的系统负担在批量处理海量数据时无法满足高吞吐率的要求另外有一些消息队列框架则被设计成实时消息处理系统虽然可以带来很高的实时处理性能但是在批量离线场合时无法提供足够的持久性即可能发生消息丢失。同时在大数据时代涌现的新的日志收集处理系统如 Flume、Scribe 等往往更擅长批量离线处理而不能较好地支持实时在线处理。相对而言Kafka 可以同时满足在线实时处理和批量离线处理。最近几年Kafka 在大数据生态系统中开始发挥着越来越重要的作用在 Uber、Twitter、Netflix、LinkedIn、Yahoo、Cisco、Goldman Sachs 等公司得到了大量的应用。目前在很多公司的大数据平台中Kafka 通常扮演数据交换枢纽的角色。传统的关系数据库一直是企业关键业务系统的首选数据库产品能够较好地满足企业对数据一致性和高效复杂查询的需求。但是关系数据库只能支持规范的结构化数据存储无法有效应对各种不同类型的数据比如各种非结构化的日志记录、图结构数据等同时面对海量大规模数据也显得“力不从心”。因此关系数据库无法达到“一种产品满足所有应用场景”的目的。在这样的大背景下各种专用的分布式系统纷纷涌现包括离线批处理系统如 MapReduce、HDFS等、NoSQL 数据库如 Redis、MongoDB、HBase、Cassandra 等、流计算框架如 Storm、S4、Spark Streaming、Samza 等、图计算框架如 Pregel、Hama 等、搜索系统如 ElasticSearch、Solr 等等。这些系统不追求“大而全”而是专注于满足企业某一方面的业务需求因此有很好的性能。但是随之而来的问题是如何实现这些专用系统与 Hadoop 系统各个组件之间数据的导入和导出。一种简单的想法是为各个专用系统单独开发数据导入导出工具。这种解决方案在技术上没有实现难度但是带来了较高的实现代价。因为每当有一款新的产品加入企业的大数据生态系统中就需要为这款产品开发和 Hadoop 各个组件的数据交换工具。因此有必要设计一种通用的工具起到数据交换枢纽的作用。当其他工具加入大数据生态系统后只需要开发和这款通用工具的数据交换方案就可以通过这个交换枢纽轻松实现和其他 Hadoop 组件的数据交换。Kafka 就是一款可以实现这种功能的产品。在公司的大数据生态系统中可以把 Kafka 作为数据交换枢纽不同类型的分布式系统如关系数据库、NoSQL 数据库、流处理系统、批处理系统等可以统一接入 Kafka从而实现和 Hadoop 各个组件之间的不同类型数据的实时高效交换较好地满足各种企业的应用需求如图5所示。同时借助于 Kafka 作为交换枢纽也可以很好地解决不同系统之间的数据生产/消费速率不同的问题。比如在线上实时数据需要写入 HDFS 的场景中线上数据不仅生成速率快而且具有突发性如果直接把线上数据写入 HDFS可能导致高峰时间 HDFS 写入失败在这种情况下就可以先把线上数据写入 Kafka然后借助于 Kafka 导入 HDFS。图5 Kafka作为数据交换枢纽小结本文介绍了 Hadoop 生态系统中三个具有代表性的功能组件。Pig 是一种数据流处理脚本语言通过提供类似 SQL 的 Pig Latin 语法让用户无需编写复杂的 MapReduce 程序即可完成海量数据的 ETL 处理大幅缩短开发周期在 Yahoo!、Twitter 等公司得到广泛应用。Tez 是基于 DAG 作业的计算框架将 Map 和 Reduce 操作细分为更细粒度的元操作并进行灵活组合通过减少不必要的 Map 阶段和 HDFS 读写次数显著提升了 Hive、Pig 等框架的执行效率性能可提升约 100 倍。Kafka 是一种高吞吐量的分布式发布/订阅消息系统能够同时满足实时在线处理的低延迟和批量离线处理的高吞吐量需求在企业大数据平台中扮演着数据交换枢纽的角色实现了 Hadoop 各组件与多种专用系统之间的数据高效流转。欢迎点赞 | 收藏⭐ | 评论✍ | 关注