Hadoop平台搭建与应用（第2版）（微课版）-教案 项目5 Hadoop生态组件的安装与使用

第4页Hadoop平台搭建与应用教案NO.5教师姓名授课班级授课形式理实一体化授课时间年月日第周授课时数16授课章节名称项目5Hadoop生态组件的安装与使用教学目的（1）识记Hadoop常用工具组件（Sqoop、Pig、Flume）的作用。（2）领会Hadoop各组件的功能与联系。（4）学会Hadoop常用工具组件（Sqoop、Pig、Flume）的使用。教材分析教学重点教学难点Hadoop常用工具组件（Sqoop、Pig、Flume）的使用更新、补充、删节内容课外作业熟练Hadoop常用工具组件（Sqoop、Pig、Flume）的安装教学过程教学提示项目

5Hadoop生态组件的安装与使用任务5.1Sqoop的安装与应用Sqoop（SQL-to-Hadoop）是一个开源工具，主要用于在Hadoop（Hive）与传统的数据库（MySQL、Oracle等）之间进行数据传递，可以将一个关系型数据库（如MySQL、Oracle、PostgreSQL等）中的数据导入到Hadoop的HDFS中，也可以将HDFS的数据导入到关系型数据库中。对于某些NoSQL数据库，Sqoop也提供了连接器。Sqoop类似于其他ETL[Extract-Transform-Load，用来描述将数据从源端经过萃取（Extract）、转换（Transform）、加载（Load）至目的端的过程]工具，使用元数据模型来判断数据类型，并在数据从数据源转移到Hadoop时确保类型进行安全的数据处理。Sqoop专为大数据批量传输设计，能够通过分割数据集并创建Hadoop任务来处理每个区块。Sqoop项目开始于2009年，最早作为Hadoop的一个第三方模块存在，后来为了让使用者快速部署，也为了使开发人员更快速地进行迭代开发，Sqoop独立成为一个Apache项目。1．Sqoop的核心功能Sqoop的核心功能包含以下2项。（1）导入数据：MySQL、Oracle导入数据到Hadoop的HDFS、Hive、HBase等数据存储系统中。（2）导出数据：从Hadoop的文件系统中导出数据到关系型数据库（MySQL等）中。Sqoop本质上是一个命令行工具。2．Sqoop中import命令的使用（1）默认情况下，通过使用“import”命令可导入数据到HDFS中。（2）指定目录和Mapper个数，并导入到HDFS中。①创建目录。${HADOOP_HOME}/bin/hdfsdfs-mkdir-p/user/sqoop/②设置Mapper个数为1，指定目录为/bin/sqoop/，如果目标目录已经存在，则先删除原目录，再创建新目录。（3）将增量数据导入到HDFS文件中时，可以通过对“check-column”“incremental”“last-value”3个参数进行设置来实现。（4）指定文件格式并导入HDFS。默认情况下，导入数据到HDFS中时，文件存储格式为textfile，可以通过对属性进行指定，以设定文件存储格式为parquet。（5）指定压缩格式并导入HDFS。默认情况下，导入的HDFS文件是不压缩的，可以通过属性“compress”“compression-codec”进行设置。（6）将Select查询结果导入到HDFS中时，必须在Where子句中包含'$CONDITIONS'。导入数据到Hive中，实例如下。$bin/sqoopimport\--connectjdbc:mysql://hostname:3306/mydb\--usernameroot\--passwordroot\--tablemytable\--num-mappers1\--hive-import\--hive-databasemydb\--hive-tablemytable\--fields-terminated-by"\t"\--delete-target-dir\--hive-overwrite3．Sqoop中export命令的使用（1）这里以数据导入为例进行说明。①导入数据到HDFS中。export--connectjdbc:mysql://hostname:3306/mydb--usernameroot--passwordroot--tablemytable--num-mappers1--export-dir/user/hive/warehouse/mydb.db/mytable--input-fields-terminated-by"\t"②执行脚本。$bin/sqoop--options-file***.opt（2）从Hive或者HDFS中导出数据到MySQL中，实例如下。$bin/sqoopexport\--connectjdbc:mysql://hostname:3306/mydb\--usernameroot\--passwordroot\--tablemytable\--num-mappers1\--export-dir/user/hive/warehouse/mydb.db/mytable\--input-fields-terminated-by"\t"任务5.2Pig的安装和使用1．ApachePig概述ApachePig是MapReduce的一个抽象。它是一个工具/平台，用于分析较大的数据集，并将数据集表示为数据流。Pig通常与Hadoop一起使用，可以使用ApachePig在Hadoop中执行所有的数据处理操作。当要编写数据分析程序时，Pig中提供了一种称为PigLatin的高级语言。该语言提供了各种操作符，程序员可以利用其开发自己的用于读取、写入和处理数据的程序。要想使用ApachePig分析数据，程序员需要使用PigLatin语言编写脚本。所有脚本都在内部转换为Map和Reduce任务。Pig的工作原理如图5-3所示。图5-3Pig的工作原理2．PigLatin的数据模型PigLatin的数据模型是完全嵌套的，它允许使用复杂的非原子数据类型，如Map和Tuple。PigLatin的数据模型。（1）Atom（原子）：PigLatin中的任何数据类型的单个值都称为Atom。它存储为字符串，可以用作字符串和数字。Int、long、float、double、chararray和bytearray都是Pig的原子值。一条数据或一个简单的原子值被称为字段，如“raja”或“30”。（2）Tuple（元组）：由有序字段集合形成的记录称为元组，字段可以是任意类型。元组与关系数据库管理系统表中的行类似，如（Raja，30）。（3）Bag（包）：一个包是一组无序的元组。每个元组可以有任意数量的字段（灵活模式）。包由“{}”表示。它类似于关系数据库管理系统中的表，但是与关系数据库管理系统中的表不同的是，包不需要每个元组中都包含相同数量的字段，或者相同位置（列）中的字段具有相同类型。例如，{（Raja，30），（Mohammad，45）}。（4）Map（映射）：映射（或数据映射）是一组Key/Value对。其中，Key必须是字符数组类型，且应该是唯一的；Value可以是任何类型，它由“[]”表示。例如，['name''Raja'，'age'30]。（5）Relation（关系）：一个关系是一个元组的包。PigLatin中的关系是无序的（不能保证按任何特定顺序处理元组）。3．Pig语句基础在使用PigLatin处理数据时，语句是基本结构。每个语句以分号（;）结尾。使用PigLatin提供的运算符可通过语句执行各种操作。除了LOAD和STORE语句之外，在执行其他操作时，PigLatin语句采用关系作为输入，并产生另一个关系作为输出。只要在Shell中输入Load语句，就会执行语义检查。要查看模式的内容，需要使用Dump运算符。只有在执行Dump操作后，才会执行将数据加载到文件系统中的MapReduce任务。PigLatin的数据类型如表5-1所示。表5-1PigLatin的数据类型序号数据类型说明＆示例1int表示有符号的32位整数。示例：82long表示有符号的64位整数。示例：5L3float表示有符号的32位浮点数。示例：5.5F4double表示有符号的64位浮点数。示例：10.55chararray表示UnicodeUTF-8格式的字符数组。示例：'w3cschool'6bytearray表示字节数组7boolean表示布尔值。示例：true/false8datetime表示日期时间。示例：1970-01-01T00:00:00.000+00:009biginteger表示JavaBigInteger。示例：6070809070910bigdecimal表示JavaBigDecimal。示例：185.9837625627289388311tuple元组是有序的字段集。示例：（raja,30）12bag包是元组的集合。示例：{（raju,30）,（Mohammad,45）}13map映射是一组键值对。示例：['name''Raju','age'30]上述数据类型的值可以为null。ApachePig以与SQL类似的方式处理空值。null可以是未知值或不存在值，它用作可选值的占位符。PigLatin的结构运算符如表5-2所示。表5-2PigLatin的结构运算符运算符描述示例()元组构造函数运算符，此运算符用于构建元组（Raju,30）{}包构造函数运算符，此运算符用于构造包{（Raju,30）,（Mohammad,45）}[]映射构造函数运算符，此运算符用于构造一个映射[nameRaja,age30]PigLatin的关系运算符如表5-3所示。表5-3PigLatin的关系运算符运算符描述加载和存储LOAD将数据从文件系统（local/HDFS）加载到关系中STORE将数据从文件系统（local/HDFS）存储到关系中过滤FILTER从关系中删除不需要的行DISTINCT从关系中删除重复行FOREACH，GENERATE基于数据列生成数据转换STREAM使用外部程序转换关系分组和连接JOIN连接两个或多个关系COGROUP将数据分组为两个或多个关系GROUP在单个关系中对数据进行分组CROSS创建两个或多个关系的向量积排序ORDER基于一个或多个字段按排序（升序或降序）排列关系LIMIT从关系中获取有限数量的元组合并和拆分UNION将两个或多个关系合并为单个关系SPLIT将单个关系拆分为两个或多个关系诊断运算符DUMP在控制台中输出关系的内容DESCRIBE描述关系的模式EXPLAIN查看逻辑、物理或MapReduce执行计划以计算关系ILLUSTRATE查看一行4．输入和输出（1）加载任何一种数据流的第一步都是指定输入。在PigLatin中，通过LOAD语句来完成输入操作。默认情况下，LOAD使用默认加载函数PigStorage，加载存放在HDFS中且以“Tab”键进行分割的文件，例如，divs=load'pig_test'。用户也可以通过指定一个完整的URL路径来加载文件，如hdfs:///data/pig_test，其表示可以从NameNode为的HDFS中读取文件。实际上，用户的大部分数据并非是使用“Tab”键作为分割符的文本文件，也有可能需要从其他非HDFS的存储系统中加载数据。Pig允许用户在加载数据时通过using句式指定其他加载函数。例如，从HBase中加载数据，语句如下。divs=load'pig_test'usingHBasestorage();如果没有指定加载函数，那么会使用内置的加载函数PigStorage。用户同样可以通过using句式为使用的加载函数指定参数。例如，如果想读取以逗号分割的文本文件数据，那么PigStorage会接收一个指定分割符的参数，语句如下。divs=load'pig_test'usingPigStorage(',');LOAD语句也可以有as句式，这个句式可以为用户加载的数据指定模式，模式相当于列名。当从HDFS中访问指定“文件”的时候，用户也可以指定目录。在这种情况下，Pig会遍历用户指定的目录中的所有文件，并将它们作为LOAD语句的输入。PigStorage和TextLoader是内置的可操作HDFS文件的Pig加载函数，是支持模式匹配的。通过模式匹配，用户可以读取不在同一个目录中的多个文件，或者读取一个目录中的部分文件。其正则匹配语法如下：？匹配任意单个字符；*匹配零个或多个字符；[abc]匹配字符集合{a,b,c}所包含的任何一个字符，[a-z]匹配指定范围内的任意字符；[^abc]匹配未包含的任意字符，其中^符号匹配输入字符串的开始位置；[^a-z]匹配不在指定范围内的任意字符；\c移除（转义）字符c所表达的特殊含义；{ab,cd}匹配字符串集合{ab,cd}中的任一字符串。（2）存储当用户处理完数据之后，需要把结果写到某个地方。Pig提供了STORE语句进行写入数据的操作。默认情况下，Pig使用PigStorage将结果数据以“Tab”键作为分割符，存储到HDFS的“storeprocessedinto'/data'”中。如果用户没有显式指定存储函数，那么将会默认使用PigStorage。用户可以使用using句式指定不同的存储函数。例如：storeprocessedinto'/data'usingHBaseStorage();用户也可以传参数给其使用的存储函数。例如，如果想将数据存储为以逗号分隔的文本数据，则PigStorage会接收一个指定分隔符的参数“storeprocessedinto'/data'usingPigStorage(',');”。当写到文件系统中后，data目录中包含多个文件，而不是一个文件。但是到底会生成多少个文件要取决于执行STORE操作前的最后一个任务的并行数（该值由为这个任务所设置的并行级别所决定）。（3）输出可以将关于结果的数据输出到屏幕上，这在调试阶段和原型研究阶段是特别有用的。dump语句可以将用户的脚本输出到屏幕上（即“dumpprocessed”）。5．PigLatin常用操作（1）查询固定行数据tmp_table_limit=limittmp_table50;dumptmp_table_limit;（2）查询指定列数据tmp_table_name=foreachtmp_tablegeneratename;dumptmp_table_name;（3）为列取别名tmp_table_column_alias=foreachtmp_tablegeneratenameasusername,ageasuserage;dumptmp_table_column_alias;（4）按某列进行排序tmp_table_order=ordertmp_tablebyageasc;dumptmp_table_order;（5）按条件进行查询tmp_table_where=filtertmp_tablebyage>18;dumptmp_table_where;（6）内连接tmp_table_inner_join=jointmp_tablebyage,tmp_table2byage;dumptmp_table_inner_join;（7）左连接tmp_table_left_join=jointmp_tablebyageleftouter,tmp_table2byage;dumptmp_table_left_join;（8）右连接tmp_table_right_join=jointmp_tablebyagerightouter,tmp_table2byage;dumptmp_table_right_join;（9）全连接tmp_table_full_join=jointmp_tablebyagefullouter,tmp_table2byage;dumptmp_table_full_join;（10）交叉查询多张表tmp_table_cross=crosstmp_table,tmp_table2;dumptmp_table_cross;（11）分组tmp_table_group=grouptmp_tablebyis_child;dumptmp_table_group;（12）分组并统计tmp_table_group_count=grouptmp_tablebyis_child;tmp_table_group_count=foreachtmp_table_group_countgenerategroup,count($1);dumptmp_table_group_count;（13）查询并去重tmp_table_distinct=foreachtmp_tablegenerateis_child;tmp_table_distinct=distincttmp_table_distinct;dumptmp_table_distinct;任务5.3Flume的安装和使用Flume作为Cloudera开发的实时日志收集系统，受到了业界的认可与广泛应用。Flume初始的发行版本目前被统称为FlumeOG（OriginalGeneration），属于Cloudera。但随着Flume功能的扩展，FlumeOG代码工程臃肿、核心组件设计不合理、核心配置不标准等缺点暴露出来，尤其是在FlumeOG的最后一个发行版本0.9.4中，日志传输不稳定的现象尤为严重，为了解决这些问题，2011年10月22日，Cloudera完成了Flume-728的开发，对Flume进行了里程碑式的改动，重构了其核心组件、核心配置及代码架构，重构后的版本统称为FlumeNG（NextGeneration）；这一改动的另一原因是需将Flume纳入Apache旗下，ClouderaFlume改名为ApacheFlume。Flume是一个分布式的、可靠的和高可用的海量日志采集、聚合和传输系统。其支持在日志系统中定制各类数据发送方，用于收集数据。同时，Flume提供对数据进行简单处理，并写到各种数据接收方（如文本、HDFS、HBase等）的功能。Flume的数据流由事件(Event)贯穿始终。事件是Flume的基本数据单位，它携带有日志数据(字节数组形式)及头信息，这些Event由Agent外部的Source生成，当Source捕获事件后，会进行特定的格式化，且Source会把事件推入到(单个或多个)Channel中。可以把Channel看作一个缓冲区，它将保存事件直到Sink处理完该事件。Sink负责持久化日志或者把事件推向另一个Source。Flume主要由以下3个重要的组件构成。（1）Source：完成对日志数据的收集，分为Transtion和Event并推入到Channel之中，Flume提供了各种Source的实现，包括AvroSource、ExceSource、SpoolingDirectorySource、NetCatSource、SyslogSource、SyslogTCPSource、SyslogUDPSource、HTTPSource、HDFSSource等。（2）Channel：主要提供队列的功能，对Source提供的数据进行简单的缓存。Flume中的Channel有MemoryChannel、JDBCChanel、FileChannel等。（3）Sink：用于取出Channel中的数据，并将其存储到文件系统、数据库或者提交到远程服务器中。其包括HDFSSink、LoggerSink、AvroSink、FileRollSink、NullSink、HBaseSink等。任务5.4Kafka的安装和使用Kafka是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，由服务器和客户端组成，通过高性能TCP网络协议进行通信。它可以部署在本地和云环境中的裸机硬件、虚拟机和容器上。服务器：Kafka作为一个或多个服务器的集群运行，这些服务器可以跨越多个数据中心或云区域。其中一些服务器形成了存储层，称为代理。其他服务器运行KafkaConnect，将数据作为事件流不断导入和导出，以将Kafka与现有系统（如关系数据库和其他Kafka集群）集成。客户端：它们允许编写分布式应用程序和微服务，即使在网络问题或机器故障的情况下，也可以并行、大规模、容错地读取、写入和处理事件流。Kafka是一个事件流平台，其以事件流的形式从数据库、传感器、移动设备、云服务和软件应用程序等事件源实时捕获数据，持久地存储这些事件流以供以后检索，实时地以及回顾性地操纵、处理和响应事件流，或根据需要将事件流路由到不同的目的地，Kafka最大的特性就是可以实时的处理大量数据以满足各种需求场景。Kafka可以与Flume、SparkStreaming、Storm、HBase、Flink和Spark协同工作，对流数据进行实时读取、分析和处理,应用实例如下。（1）实时处理支付和金融交易，例如在证券交易所、银行和保险中。（2）实时跟踪和监控汽车、卡车、车队和货运，如物流和汽车行业。（3）持续捕获和分析来自物联网设备或其他设备（如工厂和风电场）的传感器数据。（4）收集客户互动和订单并立即做出反应，例如在零售、酒店和旅游业以及移动应用程序中。（5）监测医院护理中的患者并预测病情变化，以确保在紧急情况下及时治疗。（6）连接、存储并提供由公司不同部门生成的数据。（7）作为数据平台、事件驱动架构和微服务的基础。一个事件记录了“发生的一些事情”的事实，也称为记录或消息。当读或写数据到Kafka时，是以事件的形式来完成的。从概念上讲，事件有一个键（Key）、值（Value）、时间戳（Timestamp）和可选的元数据（Metadata）头。事件示例如下。事件的Key：“佩奇”；事件的Value：“向乔治支付了100元”；事件的Timestamp：“2023年1月1日下午1:11”。生产者是那些向Kafka发布（写入）事件的客户端应用程序，消费者是那些订阅（读取和处理）这些事件的人、应用程序等。在Kafka中，生产者和消费者是完全解耦的，彼此不可知，这是实现Kafka高可扩展性的关键。例如，生产商永远不需要等待消费者。事件按主题进行组织和持久存储。主题类似于文件系统中的目录，事件是该目录中的文件。例如主题名称可以是“支付”。Kafka中的主题总是多生产者和多订阅者，即一个主题可以有零个、一个或多个向其写入事件的生产者，也可以有零、一个、或多个订阅这些事件的消费者。可以根据需要随时读取主题中的事件。与传统的消息传递系统不同，使用后不会删除事件。相反，可以通过每个主题的配置设置来定义Kafka应该保留事件多长时间，之后旧事件将被丢弃。主题是分区的，这意味着一个主题分布在位于不同Kafka代理上的多个“桶”上。这种数据的分布式放置对于可扩展性非常重要，因为它允许客户端应用程序同时从多个代理读取数据和向多个代理写入数据。当一个新事件发布到一个主题时，它实际上被附加到该主题的一个分区。具有相同Key（例如，客户或车辆ID，在图中用它们的颜色表示）的事件被写入同一分区，Kafka保证给定主题分区的任何消费者将始终以与写入时完全相同的顺序读取该分区的事件。为了使数据具有容错性和高可用性，每个主题都可以复制，甚至可以跨地理区域或数据中心复制，这样总是有多个代理拥有一份数据副本，以防出现问题，默认副本参数设置为3，即始终有三份数据副本，该复制是在主题分区级别执行的。除了用于管理和管理任务的命令行工具外，Kafka还有五个用于Java和Scala的核心API：AdminAPI：用于管理和查看主题、代理和其他Kafka对象。ProducerAPI：用于向一个或多个Kafka主题发布（编写）事件流。ConsumerAPI：用于订阅（读取）一个或多个主题并处理为其生成的事件流。KafkaStreamsAPI：用于实现流处理应用程序和微服务。它提供了处理事件流更高级别的功能，包括转换、聚合、联接等。从一个或多个主题读取输入，以便生成一个或更多主题的输出，从而有效地将输入流转换为输出流。KafkaConnectAPI用于构建和运行可重复使用的数据导入/导出连接器，这些连接器从外部系统和应用程序消费（读取）或生产（写入）事件流，以便与Kafka集成。在实践中通常不需要实现自己的连接器，因为Kafka社区已经提供了数百个现成的连接器。任务5.5Flink的安装和使用ApacheFlink是一个面向数据流处理和批量数据处理的分布式开源计算框架，它基于Flink流式执行模型（streamingexecutionmodel），能够支持流处理和批处理两种应用类型,适用于各种实时数据处理场景，可以对无界和有界数据流（有界数据流指的是有限大小的数据集，而无界数据流指的是无限的数据流，例如来自传感器、日志、消息队列等源的数据。）进行高效、准确、可靠的实时计算和数据处理。Flink的架构如图5-9所示，Flink部署(Deploy)，支持local，和集群(standalone，yarn)，也支持在云上部署。Flink的核心（core），分布式的流处理引擎。Flink的API和类库，有DataStreamAPI和DataSetAPI，分别做流处理和批处理。DataStreamAPI支持复杂事件处理，table操作和SQL操作。DatasetAPI支持FlinkML机器学习，Gelly图计算，table操作和SQL操作。Flink包含三大核心组件.DataSource，数据源(负责接收数据)；Transformations算子(负责对数据进行处理)；DataSink输出组件(负责把计算好的数据输出到其它存储介质中)。Flink系统的架构是基于Master-Slave架构。Flink集群启动时，会启动一个JobManager进程、至少一个TaskManager进程。在Local模式下，会在同一个JVM内部启动一个JobManager进程和TaskManager进程。JobManager负责接受并管理作业（Job），TaskManager负责执行任务（Task）。一个作业由一个或多个任务组成，这些任务可以分配到不同的TaskManager上运行。Ta