大数据技术及应用-基于Python语言 课件 第6、7章 Hive基础与应用、分布式计算框架MapReduce

第6章Hive基础与应用本章学习目标了解数据库与数据仓库的区别，了解Hive在Hadoop平台的位置和特征；掌握Hive存储模型，掌握Hive数据表的组织方式；掌握Hive的常用命令，能够结合数据模型理解和灵活使用DDL实现数据仓库表构建、分区设置和应用等操作；能够灵活使用DML语言，实现对Hive数据仓库表数据的操纵；掌握Hive-QL命令实现对Hive数据仓库表的构建、分区应用、数据操纵和数据查询操作；01Hive简介02Hive的存储模型目录Contents03Hive初探04Hive的数据定义—DDL05Hive的数据操纵—DML06Hive的数据检索—Hive-QL07本章小结Hive简介01Hive简介ApacheHive是由Facebook开源，用于解决海量结构日志的数据仓库软件。Hive使用类SQL（Hive-QL，HQL）语法进行数据的读取、写入、检索等操作，本质是把HQL转化为MapReduce程序。为什么要用HiveMapReduce的出现，极大的简化大数据编程的难度，使得普通程序员也能从事开发大数据编程。但数据分析人员通常使用SQL语言进行大数据统计以及分析，而Mapreduce编程是有一定的门槛，如果每次都采用MapReduce开发，则成本太高、效率太低。那么有没有更简单的办法，可以直接通过SQL在大数据平台下运行进行数据分析？有的，答案即是Hive。数据库与数据仓库对比角度数

据

库数据仓库目的目的在于记录目的在于分析应用处理方法在线事务处理(OLTP)在线分析处理(OLAP).应用范围数据库用于支持业务过程的日常基础操作数据仓库使你可以对业务进行分析表与连接数据库的表和联接是复杂的，但是它们是规范化的。数据库的表和联接比较简单，它们是非规范化的。面向面向应用的数据收集面向主题的数据收集数据库与数据仓库对比角度数

据

库数据仓库存储限制通常限制于单个的应用范围所存储数据常被应用于多个应用中可用性数据是实时可用的数据只有在需要的时候再进行更新，非实时性的模型用ER模型进行数据库的设计用数据建模技术进行模型的构建技能捕获实时在线数据分析数据数据库与数据仓库对比角度数

据

库数据仓库数据类型数据库中存储的总是最新的数据记录.当前和历史数据都存储在数据仓库，通常情况下都不是最新数据数据存储利用扁平的关系方法进行数据存储利用空间规范化方法作为数据存储结构，例如采用星型、雪花型存储模式查询类型简单的事务查询以分析数据为目的的复杂检索数据汇总存储数据的明细存储数据汇总信息Hive体系结构与接口Hive体系结构与接口元数据存储：Hive将元数据存储在数据库中(MySql或者Derby)Hive体系结构Client：用户接口组件主要包含CLI（命令行接口）、JDBC或ODBC、WEBGUI（以浏览器访问Hive）；ThriftServer：Thrift是FaceBook开发的一个软件框架，它用来进行可扩展且跨语言的服务的开发，通过该服务使不同编程语言调用Hive的接口；Hive简介Driver（执行引擎）：包括Complier、Optimizer和Executor，它们的作用是将Hive-QL语句进行语法分析、语法解析、语法优化，生成执行计划，然后提交给HadoopMapReduce计算框架处理；Metastore服务：元数据服务组件，提供HDFS中存储的数据与对应表结构的关系。Hive体系结构Hive本身主要存储数据仓库的元数据，其数据可以分布存储在Hadoop系统的各个节点中，这种模式大大减轻了中心服务器的压力，从而也为大数据处理和数据仓库数据的快速分析提供了保障。由其体系结构可以看出，Hive提供了多种数据访问方式，可以通过命令行方式直接访问，也可以通过Thrift客户端/服务器的模式进行数据的存取和访问。SQL——MapReduceHive特征Hive的数据存储在HDFS上；Hive数据分析的底层实现是MapReduce；Hive的执行程序运行在Yarn上；Hive的存储模型02Hive存储模型Hive存储模型数据库（Database）表（Table）分区（Partition）桶（Bucket）数据库（database）数据库起到命名空间的作用，主要用于避免表、视图、分区、列等基本元素的命名冲突。同时，数据库在概念上对基本元素进行隔离，所以还可为用户或用户组实施安全性定义提供支持。表（Table）表是具有相同模式的同构数据单位，但是表的具体数据存储可以有不同的方式。Hive表的概念与关系型数据库中表的概念具有相似之处。分区（partitions）每个表可以有一个或多个分区键，可用于确定数据的存储方式。分区除了作为存储单元外，还允许用户有效地识别满足指定条件的行。例如，STRING类型的date_partition和STRING类型的country_partition，其分区键的每个唯一值定义了表的一个分区例如，日期为“2009-12-23”中的所有国家为“US”的数据都是page_views表的一个分区。因此，如果只对日期为“2009-12-23”的国家为“US”的数据进行分析，则只需在表的相关分区上运行该查询，从而大大加快了分析速度。分区（partitions）需要注意的是，一个分区被命名为“2009-12-23”并不意味着该分区包含所有或仅包含该日期的数据，分区以日期命名是为了方便进行数据的组织和识别，但是Hive并不确保数据的分区组织。分区列是虚拟列，它们不是数据本身的一部分，而是在加载时派生的。这也就要求数据仓库系统设计和分区数据导入前能够对数据进行一定的检查，从而确保数据能够在逻辑上满足分区的设置。桶或集群（bucket或clusters）每个分区中的数据可以根据表中某一列的哈希函数值依次分配到桶。例如，page_views表可以由userid来分隔，userid是page_view表的一列，而不是partitions列。桶的应用会大大加快数据的检索速度。Hive初探03Hive初探Hive安装完成后，首先把$HIVE_HOME设置到当前的Hive安装目录，并把$HIVE_HOME/bin加入$PATH中。设置完成后在命令行直接运行“hive”命令启动Hive。接下来我们开始构造第一个Hive数据库。Hive初探首先，我们通过ssh连接到Hive服务器上，之后运行hive，便会进入Hive的命令行交互程序，其运行过程如下。若在本机上，则无须使用ssh连接Hive初探步骤1：构造Database/SchemaHive初探用CLI和Web方式查看所创建的数据库文件夹，其中“db_city_aq.db”虽然包含了“.db”作为后缀，但其实际上是一个文件夹。当前并未往数据库中添加任何数据表，所以文件夹为空。Hive初探采用Web方式查看数据库文件夹Hive初探步骤2：创建数据表首先利用USE命令把“db_city_aq”设置为当前的数据库，以保证我们所创建的数据表包含在该数据库中。之后，利用“CREATETABLE”创建一个数据表数据表创建完成后，利用SHOW命令查看数据库中的数据表Hive初探Hive初探步骤3：数据文件的加载，并确认Hive初探步骤4：数据的插入操作Hive初探步骤5：数据的查询，分组求平均值Hive初探本节内容完整的展示了一个Hive数据库创建、数据表创建并导入数据、以及数据查看的整个流程，希望这个简单的Hive应用过程能够让读者对Hive形成直观认识，使其不再是抽象的概念。接下来的章节中将对Hive的数据定义语言、数据操纵语言和数据查询语言逐一展开介绍。Hive的数据定义—DDL04Hive的数据定义--DDLHive的数据定义主要通过其数据定义语言（DDL：DataDefinitionLanguage）完成。DDL中的主要操作包括CREATE、DROP、SHOW、TRUNCATE、DESCRIBE、ALTER等，其操作对象可以是DATABASE，SCHEME，TABLE，VIEW等。数据库的相关操作1．创建数据库创建Database的操作命令是CREATE，其语法定义如下：CREATE(DATABASE|SCHEMA)[IFNOTEXISTS]database_name[COMMENTdatabase_comment][LOCATIONhdfs_path][MANAGEDLOCATIONhdfs_path][WITHDBPROPERTIES(property_name=property_value,...)]数据库的相关操作【例6.1】通过指定外部表存储位置的方法建立城市空气质量数据库“db_city_aq”，同时为该数据库增加说明文字。数据库的相关操作通过show查看Hive中的数据库，从查询结果中可以看到刚刚创建的数据库db_city_aq数据库的相关操作再通过HDFS文件系统上查看数据库对应的文件生成情况，利用ls查看hdfs上的文件夹发现数据库db_city_aq对应的目录数据库的相关操作2．删除数据库删除数据库的命令是DROP，其语法定义如下：DROP(DATABASE|SCHEMA)[IFEXISTS]database_name[RESTRICT|CASCADE];[RESTRICT|CASCADE]选项没有明确指定时，缺省值是RESTRICE，表示数据库不为空时不能删除数据库；若指定为CASCADE，则会级联删除数据库中的所有数据表，因此应慎重使用。数据库的相关操作【例6.2】删除空数据库db_city_aq，再利用show命令查看现有数据库，删除成功。数据库的相关操作3．使用数据库使用数据库USE命令的含义是指在交互操作过程中，使用该指令设置指定数据库为当前数据库。其语法定义如下：USEdatabase_name;USEDEFAULT;DEFAULT是缺省情况下特指某个数据库，可以用DEFAULT代替具体的DATABASE名称，以方便使用。数据表的创建Hive的数据组织采用文件夹、文件的方式进行，其数据文件存储格式可以具体指定，例如文件存储格式可以是：SEQUENCEFILE|TEXTFILE|RCFILE|ORC|PARQUET|AVRO|JSONFILE|INPUTFORMATinput_format_clsnameOUTPUTFORMAToutput_format_clsname。其中TEXTFILE为缺省文件格式，具体缺省值由hive.default.fileformat指定；JSONFILE文件格式需要Hive4.0.0及以上版本才能支持。数据表的创建内部表与外部表（ManagedandExternalTables）：缺省情况下，Hive创建的表都是内部表（也称管理表），即文件、元数据和相关统计信息由Hive进行管理和处理，内部表删除后，相关数据文件也将被删除；外部表是指所创建的数据表存放在非Hive的缺省存储位置，Hive维护指向该外部数据文件的链接，外部表常被多个应用共享，从而避免数据表被误删除。删除内部表会直接删除元数据及存储数据，删除外部表则仅仅删除元数据，HDFS上的文件并不会被删除；对内部表的修改会将修改直接同步给元数据，而对外部表的表结构和分区进行修改，则需要修复（MSCKREPAIRTABLEtable_name）。Hive基本数据类型数据表的创建1．创建数据表的核心语法定义数据表的创建【例6.4】如下创建一个内部表，并在创建完成后查看其对应的文件夹。数据表的创建2．创建带分区的数据表【例6.5】假设数据包含五列（cdate4query,chour,aq_type,city,value）为了创建基于cdate的分区表，在数据表中以"cdate4query"作为列名，确保cdate属性可以同时用于分区和查询。若在数据表中没有以"cdate4query"作为列名设置对日期进行存储，那么对该数据表进行查询时，将不会返回对应日期值。若有查询包含条件"WHEREdate='...'"，Hive将会利用分区加快查询效率，同时列"cdate4query"将会返回date的原有值。数据表的创建分区的插入数据insertoverwritetablecity_aq_with_partitionpartition(cdate='2021-01-01')values('20210303',0,'AQI','fuzhou',40);数据表的创建3．利用Select查询创建数据表Hive数据表也可以通过查询创建（create-table-as-select：CTAS）操作进行创建。由CTAS创建数据表的过程是一个执行事务的过程，无论该表有多大，数据表只有在生成所有查询结果并创建表完成后才对其他用户可见，这一执行过程对其他用户而言是透明的。数据表的创建【例6.6】利用Hive初探中的city_aq数据表为蓝本，所导入数据为全部城市空气质量数据，使用CTAS创建一个chour=10的数据表city_aq_hour10数据表的创建4．利用Like创建数据表CREATETABLE中的LIKE选项让用户可以用源数据表结构为模板创建一个具有相同表结构的新表，而不复制源数据表的数据。【例6.7】使用Like方式复制了city_aq数据表结构，新建的tbl_create_like表结构与之相同，但不复制源表中的数据。应用Like方式创建新数据表，并利用DESC查看新建数据表的结构数据表的创建数据表的创建5．创建临时表（TemporaryTables）若一个表是作为临时表创建的，那么该表只在当前会话期间可见，其数据被存储于用户的临时文件夹中，并在会话结束后删除。若临时表的表名与数据库已有的永久表的表名相同，那么在这一会话期间，用户无法访问已有的永久表，所有对同名表的访问都会被导向临时表。当临时表被删除后，才能恢复对永久表的访问指向。与永久表相比，临时表有其自身特征：（1）不支持分区；（2）不支持索引。临时表的存储位置可以通过hive.exec.temporary.table.storage配置参数设置。数据表的创建【例6.8】如下应用中创建临时数据表数据表的创建6．表的约束Hive包括未验证的主键和外键约束的支持。当存在约束时，一些SQL工具将会生成更加高效的查询。同时，由于这些约束未经验证，因此相关系统需要确保数据被载入Hive前的完整性。对Hive进行数据检索时，Hive的约束将会被用于提高Hive数据的查询优化，从而提高数据检索的效率。列约束定义包含主键约束、唯一值约束、非空约束等，相关约束及其定义如下：column_constraint_specification:[PRIMARYKEY|UNIQUE|NOTNULL|DEFAULT[default_value]|CHECK[check_expression]ENABLE|DISABLENOVALIDATERELY/NORELY]数据表的创建【例6.9】创建表tbl_pk，其中包含(id1,id2)共同组成该表的primarykey。数据表的创建【例6.10】创建表tbl_fk，其中包含外键约束关联到tbl_pk表数据表和分区的修改通过ALTER命令可以修改现有数据表的结构，具体包括增加数据列、增加分区、修改SerDe、增加数据表属性、修改表名等等。类似的，利用ALTER可以修改数据表的分区属性。数据表和分区的修改1．数据表的修改用法1修改数据表名通过ALTERTABLE操作命令，可以修改数据表的名称。数据表更名操作不但会修改数据表在Hive中的元数据信息，也将更改该数据表在HDFS中的存储位置。ALTERTABLEtable_nameRENAMETOnew_table_name;数据表和分区的修改用法2修改数据表的属性通过ALTERTABLE操作命令可以修改数据表的属性，其中数据表的定义为(property_name=property_value,...)，因此可以一次性更改多个数据表的属性值。当然，也可以为数据表增加新的属性。所有数据表的扩展属性都可以通过“DESCRIBEEXTENDEDTABLE”查看。ALTERTABLEtable_nameSETTBLPROPERTIES(‘db_pro1’=’pro1_value’);数据表和分区的修改用法3修改数据表说明通过ALTERTABLE操作命令，可以修改/设置数据表的说明信息。ALTERTABLEtable_nameSETTBLPROPERTIES('comment'=new_comment);数据表和分区的修改2．修改数据表约束用法1修改数据表的主键约束（PRIMARYKEY）ALTERTABLEtable_nameADDCONSTRAINTconstraint_namePRIMARYKEY(column,...)DISABLENOVALIDATE;数据表和分区的修改【例6.16】如下操作先创建一个数据表，之后利用ALTERTABLE增加主键约束数据表和分区的修改用法2修改数据表的外键约束（FOREIGNKEY）ALTERTABLEtable_nameADDCONSTRAINTconstraint_nameFOREIGNKEY(column,...)REFERENCEStable_name(column,...)DISABLENOVALIDATERELY;用法3修改数据表的唯一性设置（UNIQUE）ALTERTABLEtable_nameADDCONSTRAINTconstraint_nameUNIQUE(column,...)DISABLENOVALIDATE;用法4修改数据表的非空值约束（NOTNULL）ALTERTABLEtable_nameCHANGECOLUMNcolumn_namecolumn_namedata_typeCONSTRAINTconstraint_nameNOTNULLENABLE;数据表和分区的修改用法5修改数据表的缺省值约束（DEFAULT）ALTERTABLEtable_nameCHANGECOLUMNcolumn_namecolumn_namedata_typeCONSTRAINTconstraint_nameDEFAULTdefault_valueENABLE;用法6修改数据表的检查约束（CHECK）ALTERTABLEtable_nameCHANGECOLUMNcolumn_namecolumn_namedata_typeCONSTRAINTconstraint_nameCHECKcheck_expressionENABLE;用法7删除数据表的约束（DROPCONSTRAINT）ALTERTABLEtable_nameDROPCONSTRAINTconstraint_name;数据表和分区的修改3．修改分区设置通过ALTERTABLE的PARTITION子句可以对数据表的分区进行修改、增加、改名、移动、删除、存档等操作。要使元存储感知添加到HDFS的分区，可以使用元存储检查命令（MSCK）进行数据刷新。数据表和分区的修改用法1:添加新分区使用下述ALTERTABLEADDPARTITION指令可以增加数据表的分区。其中的分区值是字符串类型时，需要用单引号包括，分区存储位置必须是数据文件所在文件夹。ADDPARTITION会修改数据表的元数据，但是不会实现已有数据的重新分区，如果在指定分区位置不存在对应数据，那么查询将不会返回任何结果。ALTERTABLEtable_nameADD[IFNOTEXISTS]PARTITIONpartition_spec[LOCATION'location'][,PARTITIONpartition_spec[LOCATION'location'],...];partition_spec:(partition_column=partition_col_value,...)数据表和分区的修改【例6.23】如下实例首先创建一个page_view数据表，该数据表设置了基于日期dt和国家country的分区。之后使用ALTERTABLEADDPARTITION指令在数据表中手动添加多个分区，并设置其对应的文件夹。下列只列出一种用法，其余可参考教材。数据表和分区的修改数据表和分区的修改4．修改数据表/分区本小节上述内容介绍了适用于修改数据表和修改数据分区的操作命令，这里介绍的操作指令既适用于数据表的修改，也适用于分区的修改。数据表和分区的修改用法1修改表/分区文件格式该命令用于修改数据表/分区的文件存储格式，其可用的file_format选项可以参考CREATETABLE中的文件类型。该命令的修改仅仅作用于数据表的元数据中，对文件的实际数据格式修改需要在Hive外进行数据转换。具体命令如下：ALTERTABLEtable_name[PARTITIONpartition_spec]SETFILEFORMATfile_format;用法2修改表/分区存储位置ALTERTABLEtable_name[PARTITIONpartition_spec]SETLOCATION"newlocation";数据表和分区的修改用法3修改表/分区的保护机制对数据的保护可以设置于表级别，也可以设置于分区级别。NO_DROP选项将会确保数据表不被删除，OFFLINE选项则使得数据表/分区中的数据被检索，而他们的元数据仍可被存取访问。若某个分区被设置为NO_DROP，那么该分区所在的表也将是不可删除的；相反，若数据表是NO_DROP的，其包含的分区是可删除的；若数据表设置为NO_DROPCASCADE，那么其分区不能被删除；若要强制删除分区，则可以使用IGNOREPROTECTION选项。修改数据表/分区的保护机制的操作命令如下。ALTERTABLEtable_name[PARTITIONpartition_spec]ENABLE|DISABLENO_DROP[CASCADE];ALTERTABLEtable_name[PARTITIONpartition_spec]ENABLE|DISABLEOFFLINE;数据表和分区的修改5．修改数据列注意：Hive中的列名是大小写敏感的用法1修改列名/类型/存储位置/说明如下命令允许用户修改列名、列数据类型、说明文字、存储位置，或者上述属性的组合情况。CASCADE关键字使得数据表元数据中的列属性同时传递变化到所有分区元数据中。RESTRICT为缺省值，限制列的改变只作用于表的元数据。ALTERTABLEtable_name[PARTITIONpartition_spec]CHANGE[COLUMN]col_old_namecol_new_namecolumn_type[COMMENTcol_comment][FIRST|AFTERcolumn_name][CASCADE|RESTRICT];数据表和分区的修改【例6.26】如下实例首先创建一个简单的数据表，之后对其数据表的属性进行修改，并调整位置数据表和分区的修改用法2：添加/替换数据列需要添加/替换数据列时，可以使用如下操作命令：ALTERTABLEtable_name[PARTITIONpartition_spec]ADD|REPLACECOLUMNS(col_namedata_type[COMMENTcol_comment],...)[CASCADE|RESTRICT]数据表的其他操作1．表的删除DROPTABLE用于删除数据表以及相关元数据，其中数据表会被移除到垃圾箱中（系统设定垃圾箱的前提下），而元数据则直接从Hive中删除。若在删除数据表时指定了PURGE参数，则会把数据文件直接清除，不可恢复。删除数据表的语法如下：DROPTABLE[IFEXISTS]table_name[PURGE];数据表的其他操作2．表的截断与删除表不同，截断表用于删除表或者分区中的全部数据，若设定了垃圾箱，则数据会被移除到垃圾箱中，否则直接删除。被截断的数据表要求为内部表，用户也可以通过partition_spec来截断特定的分区，移除指定的分区。缺省情况下，TRUNCATE将会删除所有数据，其定义如下：TRUNCATE[TABLE]table_name[PARTITIONpartition_spec];partition_spec::(partition_column=partition_col_value,partition_column=partition_col_value,...)数据表的其他操作3．SHOW命令SHOW提供了一种查看Hive中数据的元存储信息及元数据的方法。SHOW命令的详细用法请查阅教材数据表的其他操作4．Describe命令DESCRIBE命令也是用于显示数据库、表等基本元素的相关信息。但是与SHOW不同，DESCRIBE更加侧重从更高的层次上查看总体数据情况。DESCRIBE命令的详细用法请查阅教材。Hive的数据操纵—DML05Hive的数据操纵--DMLHive的数据操纵主要通过其数据操作语言（DML：DataManipulationLanguage）完成，DML中包含有LOAD、INSERT、UPDATE、DELETE、IMPORT/EXPORT、…等。下面将对其中一些常用命令进行介绍。Hive的数据操纵--DML1．加载数据文件到表（Load）Hive将数据文件加载到表时不进行任何转换，其加载操作是一种复制/移动操作，即将数据文件移动到与Hive表对应的位置。其语法格式如下：LOADDATA[LOCAL]INPATH'filepath'[OVERWRITE]INTOTABLEtablename[PARTITION(partcol1=val1,partcol2=val2...)][INPUTFORMAT'inputformat'SERDE'serde']Hive的数据操纵--DML2．把检索结果存入表（Insert）除了可以直接把文件加载到表中，Hive还提供把检索结果插入表的操作（Insert），该操作的语法如下：语法格式1：INSERTOVERWRITETABLEtablename1[PARTITION(partcol1=val1,partcol2=val2...)[IFNOTEXISTS]]select_statement1FROMfrom_statement;Hive的数据操纵--DML2．把检索结果存入表（Insert）Hive扩展2(多表插入语句):FROMfrom_statementINSERTINTOTABLEtablename1[PARTITION(partcol1=val1,partcol2=val2...)]select_statement1[INSERTINTOTABLEtablename2[PARTITION...]select_statement2][INSERTOVERWRITETABLEtablename2[PARTITION...[IFNOTEXISTS]]select_statement2]...;Hive的数据操纵--DML3．把检索结果导出到文件（Insert）Hive的检索结果可以导出到文件系统中保存成文件，其具体命令与Insert到数据表的命令应用类似，其语法定义如下：标准语法：INSERTOVERWRITE[LOCAL]DIRECTORYdirectory1[ROWFORMATrow_format][STOREDASfile_format]SELECT...FROM...Hive的数据操纵--DML3．把检索结果导出到文件（Insert）Hive的检索结果可以导出到文件系统中保存成文件，其具体命令与Insert到数据表的命令应用类似，其语法定义如下：Hive扩展语法：FROMfrom_statementINSERTOVERWRITE[LOCAL]DIRECTORYdirectory1select_statement1[INSERTOVERWRITE[LOCAL]DIRECTORYdirectory2select_statement2]...Hive的数据操纵--DML4．利用SQL的INSERTINTO把数据插入表除了支持将数据文件以及Hive检索结果数据插入表外，Hive还支持SQL的INSERTINTO实现Hive数据插入。其语法定义如下：INSERTINTOTABLEtablename[PARTITION(partcol1[=val1],partcol2[=val2]...)]VALUESvalues_row[,values_row...]values_row=(value[,value...])Hive的数据操纵--DML5．数据更新操作（Update）Hive表的数据更新操作仅可用于支持ACID的表。其语法定义如下：UPDATEtablenameSETcolumn=value[,column=value...][WHEREexpression]Hive的数据操纵--DML6．数据删除操作（Delete）与Update操作一样，Delete只能应用于支持ACID的表中。其语法格式为：DELETEFROMtablename[WHEREexpression]Hive的数据操纵--DML7．数据合并操作（Merge）合并（MERGE）操作是指利用与源数据表的连接结果更新目标数据表的操作，该操作也仅可用于支持ACID的表上，其语法格式如下：MERGEINTO<targettable>ASTUSING<sourceexpression/table>ASSON<booleanexpression1>WHENMATCHED[AND<booleanexpression2>]THENUPDATESET<setclauselist>WHENMATCHED[AND<booleanexpression3>]THENDELETEWHENNOTMATCHED[AND<booleanexpression4>]THENINSERTVALUES<valuelist>Hive的数据检索—Hive-QL06Hive的数据检索—Hive-QL在完成本小结内容前，我们新增两个数据表到数据库中，分别对应空气质量检测站点信息，以及站点采集数据的原始数据，其建表和数据导入过程请参考教科书。上述过程完成后，将会建立连个数据表（tbl_sites，tbl_site_data），并且导入一定数量的数据。Hive的数据检索—Hive-QLHiveQL提供数据检索的功能。其具体语法如下，H-QL中还提供了部分特有的检索功能，我们将结合该操作命令的语法进行介绍。SELECT[ALL|DISTINCT]select_expr,select_expr,...FROMtable_reference[WHEREwhere_condition][GROUPBYcol_list][ORDERBYcol_list][CLUSTERBYcol_list|[DISTRIBUTEBYcol_list][SORTBYcol_list]][LIMIT[offset,]rows]Hive的数据检索—Hive-QL（1）SELECTFROM关于SELECT检索，其中的FROMtable_reference表示查询操作的输入表，这些输入除了一般的数据表外，还可以是一个视图、一个join构造，或者一个子查询。需要注意的是，Hive查询语句中的表名和列名是大小写相关的。Hive的数据检索—Hive-QL如下检索只返回PM2.5的相关测量数值，并指定了返回列。Hive的数据检索—Hive-QL（2）WHEREWHERE表示查询条件，其后连接的是一个布尔表达式，例如如下查询将会返回数据表tbl_sites中所有经度大于130的检测站点。Hive的数据检索—Hive-QL在Hive中，WHERE子句中还支持子查询的应用。WHERE中的子查询需要结合IN/NOTIN，EXISTS/NOTEXISTS使用，IN/NOTIN需要选择一个单个的数据列，用来查看数据是否在所选择列的数据集中；而EXTSTS/NOTEXISTS则无须选择单个数据列，其判断依据是子查询返回结果是否为空。下面是一个简单的带子查询的检索，该检索选择所有站点经度大于100度区域的站点监测数据，并输出站点监测数据、站点编号等信息。Hive的数据检索—Hive-QLHive的数据检索—Hive-QL（3）ALL和DISTINCTALL和DISTINCT用于说明返回值中重复行的数据处理方式。ALL为缺省情况，其表示返回所有数据行；而DISTINCT则需要进一步过滤，消除重复行数据。（4）基于分区的检索一般情况下，SELECT查询会扫描整个数据表。若一个Hive的数据表包含分区，那么SELECT检索有可能无须检索整个数据表，而只检索数据表的部分分区便可完成查询操作。Hive支持分区探测，也就是说会利用分区谓词与WHERE子句，或者与JOIN中的ON子句进行综合判断，从而避免没有必要的分区数据检索。Hive的数据检索—Hive-QL（5）GROUPBY与HAVING子句GROUPBY子句用于设定检索数据分组依据，该子句通常用于查询聚合数据结果，例如查询数据的平均值、求和、或者求分组元素个数等。GROUPBY子句语法定义如下，其分组表达式用于设定具体的分组依据，通常情况下是数据列名，或者列名的表达式。groupByClause:GROUPBYgroupByExpression(,groupByExpression)*Hive的数据检索—Hive-QL（6）ORDERBY与LIMIT子句ORDERBY子句用于指定返回结果排序依据，其语法定义如下：orderBy:ORDERBYcolName(ASC|DESC)?(NULLSFIRST|NULLSLAST)?(colName(ASC|DESC)?(NULLSFIRST|NULLSLAST)?)*Hive的数据检索—Hive-QL上述只列出了Hive-QL部分最主要结构及部分检索应用实例，更加详细的操作实例可以参考教材。本章小结07本章小结Hive是基于Hadoop的一个数据仓库工具，用来进行数据提取、转化、加载，是一种可以存储、查询和分析存储在Hadoop中的大规模数据的工具。本章简要介绍了数据库管理系统与数据仓库的区别，介绍了Hive的存储模型，并以一个简单的实例展示Hive建库、建表、数据导入和查询的全过程。PPT模板下载：/moban/行业PPT模板：/hangye/节日PPT模板：/jieri/PPT素材下载：/sucai/PPT背景图片：/beijing/PPT图表下载：/tubiao/优秀PPT下载：/xiazai/PPT教程：/powerpoint/Word教程：/word/Excel教程：/excel/资料下载：/ziliao/PPT课件下载：/kejian/范文下载：/fanwen/试卷下载：/shiti/教案下载：/jiaoan/字体下载：/ziti/

谢谢！

第7章分布式计算框架MapReduce本章学习目标掌握MapReduce的基本工作原理；掌握MapReduce各步骤输入/输出的衔接关系；掌握利用HadoopStreaming进行代码测试的基本思路与方法；掌握利用Python设计MapReduce程序的思路和方法；掌握MapReduce程序设计模式的基本概念，以及理解常见的几种设计模式；理解MRJob的工作原理，并能够进行简单的MRJob程序设计应用。01MapReduce概述02WordCount实例详解目录Contents03Python代码测试04利用Python的迭代器和生成器优化wordCount程序05MapReduce程序设计模式06用MRJob库编写MapReduce程序07本章小结MapReduce概述01MapReduce概述HadoopMapReduce是一个计算框架，目的是为了方便编写具有高可靠性、高容错性的，能在大型集群（数千个节点）上并行地处理大量数据（数TB的数据集）的分布式应用程序。该计算框架管理数据传递的所有细节，如发起任务、验证任务完成情况以及在节点之间复制数据等。第一个MapReduce的Python程序Hadoop自带有计算圆周率PI的示例jar包，程序用随机数的方式计算PI值，该程序是一个java程序，通过执行以下命令调用集群进行计算：hadoopjar$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.0.jarpi1010输出结果如下：第一个MapReduce的Python程序第一个Python程序也是实现PI的计算。第一个MapReduce的Python程序第一个MapReduce的Python程序在HDFS上创建一个空的文件，之后以该空文件作为输入文件便可以运行计算PI的程序了。运行完成后，可以在HDFS文件系统中查看输出文件内容。假设把程序对应的mapper和reducer存放在“/codes/Pi”文件夹中，HDFS系统上用作输入的数据存放于HDFS的“/data”文件夹，输出存放于“/data_rlt”。其操作步骤具体如下：第一个MapReduce的Python程序chmod+x/codes/Pi/*.*第一个MapReduce的Python程序利用如下命令运行该程序运行结果如下：mapredstreaming-input/data/tmp.txt-output/data_rlt-mapper/codes/pi/pi_mapper.py-reducer/codes/pi/pi_reducer.py第一个MapReduce的Python程序MapReduce工作原理MapReduce程序分三个阶段执行，即map阶段、shuffle阶段和reduce阶段其中shuffle可以由MapReduce框架自动完成。因此，多数情况下，MapReduce算法主要包括两个任务：Map和Reduce。MapReduce工作原理1．Map阶段Map的任务是对输入数据进行处理，通常情况下，MapReduce作业的输入和输出都存储在文件系统中，MapReduce作业将输入数据集拆分为独立的块，这些块由Map任务以完全并行的方式处理，并转换为一种（key,value）的数据模式。MapReduce工作原理2．Reduce阶段该阶段包括shuffle和reduce，Reducer的任务是把Mapper生成的（key,value）集整合处理，并输出最终结果到HDFS中。系统对map任务的输出执行排序和转换，并映射为reduce任务的输入，此过程就是Shuffle。两阶段任务完成后，集群收集并缩减数据以形成最后结果，并将其发送回Hadoop服务器。MapReduce工作原理CombinerMapReduce程序还可以包含combiner，combiner在Map之后执行，通常情况下实现对Map输出数据的简单归纳，从而减少数据传输。在多数应用中，combiner都是能够大幅提高程序效率的关键步骤。MapReduce工作原理Hadoop框架是用Java实现的，MapReduce应用程序却可以不用Java编写。Hadoop流（HadoopStreaming）是一个工具应用程序（Utility），它允许用户使用任何可执行文件作为mapper或reducer来创建和运行作业；此外，Hadoop管道（HadoopPipes）是一种SWIG兼容的C++API，他可以支持编写非JNI（JavaNativeInterface）的MapReduce应用程序。MapReduce作业的工作流程MapReduce作业的工作流程用户程序中的MapReduce库首先将输入文件划分为M片，每片大小一般在16M到64M之间（可由参数指定），然后，它在集群多台机器上启动相同的程序复制。其中一个复制程序为master(主模块)，其余的都是worker(分模块)。worker接收master分配的任务，其中有M个Map任务和R个Reduce任务要分配。master挑选一个空闲的worker并为其分配一个map任务或者reduce任务。MapReduce作业的工作流程被分配到Map任务的worker会去读取相应的输入块的内容。它从输入文件中解析出键值对并且将每个键值对传送给用户定义的Map函数。由Map函数产生的中间键值对则缓存在内存中。被缓存的键值对会阶段性地写回本地磁盘，并且被划分函数分割成R份。这些缓存在磁盘上的数据位置会被回传给master(主模块)，master再负责将这些位置转发给Reduceworker。MapReduce作业的工作流程当Reduceworker(Reduce分模块)从master(主模块)那里接收到这些位置信息时，它会使用远程过程调用从Mapworker的本地磁盘中获取缓存的数据。当Reduceworker读入全部的中间数据之后，将根据中间键对进行排序，这样所有具有相同键的键值对就都聚集在一起。因为要把具有相同键的键值对映射到同一个reducetask中，所以需要对中间键值对进行排序。MapReduce作业的工作流程Reduceworker遍历已经排好序的中间数据，每当遇到一个新的中间键，它会将key和相应的中间值传递给Reduce函数。Reduce函数的输出会被添加到这个Reduce部分的输出文件中。当所有的Maptasks和Reducetasks都完成时，master将唤醒用户程序。至此，用户代码中的MapReduce调用返回。MapReduce作业的工作流程执行完后，MapReduce的执行结果被存放在R个输出文件中（每个Reducetask对应一个，文件名由用户指定）。因为这些输出文件经常作为另一个MapReduce调用的输入，或者将它们用于另外一个能够以多个文件作为输入的分布式应用，所以，用户并不需要将R个输出文件归并成一个。MapReduce作业的工作流程MapReduce的输入/输出MapReduce框架只对<key，value>（键值对）进行操作，该框架将作业的输入视为一组<key，value>，并生成<key，value>作为作业的输出。与Java程序不同，Python的Map程序输入/输出流为标准输入/输出，STDIN和STDOUT，mapper输入从STDIN中接收数据，reducer输出结果到STDOUT；在mapper和reducer之间的数据也是通过标准输入/输出进行关联。至于他们之间的数据流关系则交由hadoopStreaming处理。WordCount实例详解02WordCount程序源码使用Python编写MapReduce的要点在于利用Hadoop流的API，通过STDIN(标准输入)、STDOUT(标准输出)在Map函数和Reduce函数之间传递数据。我们需要做的是利用Python的sys.stdin读取输入数据，并把输出传送给sys.stdout。Hadoop流将会帮助我们处理中间的过程。WordCount程序一共包含两个部分，分布式map阶段程序wordCount_mapper.py和reduce阶段程序wordCount_reducer.py，分别如下所示。WordCount程序源码WordCount程序源码WordCount程序源码WordCount程序源码编写完代码，并把他们分别存储在文件wordCount_mapper.py和wordCount_reducer.py。检查上述文件是否具有可执行权限，若没有，可以通过chmod命令设置两个文件的执行权限，如“chmod+x./wordCounter/mapper.py”。WordCount程序执行利用HadoopStreaming执行一个MapReduce任务的调用方法如下所示。hadoop-streaming.jar通常命名为hadoop-streaming-x.x.x.jar，其中x.x.x为具体版本号（例如对应hadoop3.3.0版的文件名为hadoop-streaming-3.3.0.jar）。WordCount程序执行（1）数据准备（路径可能会因个人不同）后在HDFS上创建“/wordCountData”文件夹，并把数据文件（文本文件）上传到该文件夹；WordCount程序执行（2）执行命令如下（当命令在一行中写不下时，可以使用“\”换行）mapredstreaming\-input/wordCountData\-output/wordCount_rlt\-mapper"/codes/wordCount/wordCount_mapper.py"\-reducer"/codes/wordCount/wordCount_reducer.py"WordCount程序执行（3）运行完成后，查看文件夹“/wordCount_rlt”中的文件，可以通过cat直接查看WordCount程序原理将文件拆分成多片段（splits），由于测试用的文件较小，所以每个文件为一个split，并将文件形成标准输入流，这一步由MapReduce框架自动完成；在mapper中按行读取标准输入数据，并对单词进行分割形成<key,value>，其中key为单词，value为该单词出现次数，这里没有对单词进行统计，所以value都是1；得到map方法输出的<key,value>集后，shuffle将它们按照key值进行排序，并作为reduce的输入数据；得到排过序的键值对后，利用自定义的reduce方法进行处理，得到新的<key,value>对，并作为WordCount的输出结果。WordCount程序原理MapReducePython代码的测试03MapReducePython代码的测试使用Python编写完成的MapReduce程序，通常需要进行调试、测试，以排除隐藏的逻辑问题。对运行于分布式环境的MapReduce程序而言，其运行的整个过程由mapReduce框架自动管理，在这种情况下要通过一遍遍运行查找程序，再在输出的日志中发现逻辑错误、甚至语法问题将是非常耗时且烦琐的。因此，我们需要通过使用一些工具和方法来协助程序代码的调试。MapReducePython代码的测试Python程序的运行是通过HadoopStreaming工具进行数据流的控制，并且Python的Map程序是通过标准输入STDIN接收输入数据的，所以可以通过构造标准输出作为map程序的输入，然后利用linux的管道把map输出到sort工具实现排序；排序完成后的数据再通过管道输出到reduce程序，从而实现简单测试环境的构建。而reduce的最终输出结果，则可以通过重定向标准输出到文件的方式实现结果的输出和存储。MapReducePython代码的测试MapReducePython代码的测试以前面的wordCount程序为例，可以先把输入文件之一复制到与mapper和reducer同一文件夹中，此时，该文件夹包含wordCount_mapper.py、wordCount_reducer.py两个程序文件，运行如下命令：MapReducePython代码的测试还可以把上述节点分开。例如，可以首先把mapper的输出重定向到tmp.txt文件中，以便查看mapper阶段的运行结果。然后再把中间结果作为reducer的输入便可以分步查看两个阶段的运行结果。运行命令如下：catinput.txt|./wordCount_mapper.py>tmp.txtcattmp.txt|sort-k1,1|./wordCount_reducer.py>rlt.txt利用迭代器和生成器优化wordCount程序04Python中的迭代器迭代器是访问集合元素的一种方式，它提供了一个统一的访问集合的接口。迭代器对象从集合的第一个元素开始访问，直到访问完所有的元素结束。迭代器只提供了next方法，所以只能往前不会后退。Python中的迭代器一个简单的迭代器及使用Python中的迭代器把一个类作为一个迭代器使用需要在类中实现两个方法__iter__()与__next__()Python中的生成器带有yield的函数在Python中被称为generator（生成器），生成器是迭代器的一种。例如：Python中的生成器简单而言，yield的作用就是把一个函数变成一个generator生成器常被用于数据文件读取，生成器可以根据需要读取数据。如果直接对文件对象调用read()方法，会导致不可预测的内存占用。使用yield利用固定长度的缓冲区来不断读取文件内容，我们无须编写读文件的迭代类，就可以轻松实现文件读取。Python中的生成器读取文件的生成器，每次读取BLOCK_SIZE大小itertools模块itertools是Python内置的模块，itertools包提供了更加灵活的生成循环器的工具，其使用简单且功能强大。该模块中的部分内容，可以参考教材7.6.3.其中，gropby()迭代器能够把可迭代的输入数据按关键字进行分组后输出，例如：itertools模块fromitertoolsimportgroupbyfromoperatorimportitemgettermapout=[('Hello',1),('hadoop',1),('Bye',1),('hadoop',1),('Hello',1),('world',1),('Bye',1),('world',1)]mapout=sorted(mapout,key=itemgetter(0))#排序rlt_groupby=groupby(mapout,key=itemgetter(0))forkey,itemsinrlt_groupby:print("key:",key)forsubiteminitems:print("subitem:",subitem)优化wordCount程序利用上述Python的生成器以及itertools模块工具，我们对wordCount程序进行修改，形成更具有Python风格的wordCount。优化wordCount程序优化wordCount程序#!/usr/bin/envpython#-*-coding:UTF-8-*-fromoperatorimportitemgetterfromitertoolsimportgroupbyimportsysdefread_mapper_output(file,separator='\t'):forlineinfile:yieldline.rstrip().split(separator,1)defmain(separator='\t'):data=read_mapper_output(sys.stdin,separator=separator)#data=sorted(data,key=itemgetter(0))forcurrent_word,groupingroupby(data,itemgetter(0)):try:total_count=sum(int(count)forcurrent_word,countingroup)print("%s%s%d"%(current_word,separator,total_count))exceptvalueError:passif__name__=="__main__":main()例7.6.4测试结果MapReduce程序设计模式05MR程序设计模式MapReduce程序设计模式的目的是通过总结MapReduce任务的模式，形成具有一定可替代性的程序设计模式，以便帮助程序员提高学习、实现MapReduce框架上的程序开发的效率。数据集介绍为了能够更好的示范程序的数据处理与应用，我们采用空气质量数据作为本章处理的目标数据，全国空气质量数据来自中国环境监测总站的全国城市空气质量实时发布平台。数据格式：城市空气指标值、站点空气指标、站点详细信息20140513,0,AQI,北京,8120140513,0,AQI,天津,6920140513,0,AQI,石家庄,12520140513,0,PM.5,唐山,82(1)聚合查询模式聚合查询是指对所要处理的数据进行一定的统计聚合处理，以从概要、全局的层面上掌握数据的特征和状况。例如，今年北京地区的空气质量总体情况，包括空气质量各个指标的平均值、最大最小值，或者空气质量评价为优的天数等。聚合查询模式的核心思想是按照一定的关键字对数据进行分组，然后对分组数据进行数据的聚合、整理、处理。该模式又可以具体分为数值概要模式、倒排序索引模式、计数器模式等。聚合查询模式—数值概要模式数值概要模式的目的是按照某些键值对数据进行分组，然后计算特定分组下数据的聚合值，如平均值、最大最小值、求和等。任务1：统计每个城市的PM2.5的最大值、最小值和平均值聚合查询模式--数值概要模式聚合查询模式—倒排序索引模式任务1的源代码请参考：教材P238-239mapredstreaming\-input/CityAQ/*\-output/AQI_rlt\-mapper/codes/AQI/-mapper.py\-reducer/codes/AQI/-reducer.py聚合查询模式—数值概要模式任务2：增加combiner，实现更加高效的数据统计工作；聚合查询模式—倒排序索引模式任务3的源代码请参考：教材P240但该版教材的源代码有点小bug聚合查询模式—倒排序索引模式倒排索引源于实际应用中需要根据属性的值来查找记录，这种索引表中的每一项都包括一个属性值和具有该属性值的记录的地址。由于不是由记录确定属性值，而是由属性值确定记录的位置，因而称为倒排索引(invertedindex)。带有倒排索引的文件称为倒排索引文件，简称倒排文件(invertedfile)。聚合查询模式—倒排序索引模式任务1：构建所有城市指定年份空气质量为优的日期倒排序索引。输入：城市空气质量监测记录输出：一个倒排序文件，每一条记录格式为<城市,年份,空气质量为优的日期列表>为了完成该任务，在Mapper中需要把城市空气质量符合条件的对应的日期和时刻加以记录，并输出。在Reducer中，其接受到的数据序列中，具有相同城市的数据相邻，所以只要把具有相同关键字的键值对的值取出，并形成列表便可以得到空气质量为优的日期序列。过滤模式过滤模式（FilterPattern）或称标准模式（CriteriaPattern）是一种设计模式，这种模式允许开发人员使用不同的标准来过滤一组对象，通过逻辑运算以解耦的方式把它们连接起来。这种类型的设计模式属于结构型模式，它结合多个标准来获得单一标准。过滤模式最重要的特点是保持数据的完整性，只是对记录保留与否进行判断，输出满足条件的记录。过滤模式--过滤器模式过滤器的功能就是通过设置一定的条件筛选并输出满足条件的记录，其作用类似于SQL语句中的过滤条件。在过滤器模式的实现中，由于读取的记录中已经包含了需要的全部信息，所以简单的过滤模式一般情况下都只需要mapper程序。很多情况下，若最终得到的数据是原来数据的子集，并且其体量也大大缩小，那么我们希望这些数据能够输出到一个文件中，而不是被分为多个part进行存储，这时可以利用reduce阶段的排序，再把所有数据通过一个reducer输出到一个文件中。若是一些综合性任务，这里的reducer还可能会在过滤的基础上完成其他任务。过滤模式--过滤器模式任务1：空气质量优秀的数据检索。小王是个运动爱好者，他几乎每天早上6-7点之间都进行户外运动，为此，他想看看北京市这几年在他运动期间的空气质量为“优”的详细数据，请帮他从文件中找出这些数据。输入：所有城市所有日期的空气质量状况及其各项指标；输出：北京市早上六点时空气质量为优的数据记录。过滤模式--过滤器模式过滤模式--TopK模式TopK模式是指按照一定的排序条件，从所有符合条件的数据中挑选出排在最前的K个元素并输出。对于该模式，其基本处理思路是从所有mapper中输出该节点数据的TopK，之后再reducer中对所有mapper的TopK进行汇总，输出最终的K的元素。过滤模式--TopK模式任务1：空气质量的比较。每年，大众都会对哪些城市是空气质量最好的城市争论不休，小刘老师提了个衡量城市质量的好方法，他的方法是从全国所有城市中找出空气质量最好的1000条记录，再看看这1000条记录都属于哪