Hadoop平臺搭建與應(yīng)用（第2版）（微課版） 課件 項(xiàng)目8 Hadoop平臺應(yīng)用綜合案例

任務(wù)8.1本地?cái)?shù)據(jù)集上傳到數(shù)據(jù)倉庫Hive任務(wù)實(shí)施下面把test.txt中的數(shù)據(jù)導(dǎo)入到數(shù)據(jù)倉庫Hive中。為了完成這個(gè)操作，需要先把test.txt上傳到HDFS中，再在Hive中創(chuàng)建一個(gè)外部表，完成數(shù)據(jù)的導(dǎo)入。1．啟動(dòng)HDFSHDFS是Hadoop的核心組件，因此，要想使用HDFS，必須先安裝Hadoop。這里已經(jīng)安裝了Hadoop，打開一個(gè)終端，執(zhí)行命令“start-all.sh”，啟動(dòng)Hadoop服務(wù)，操作命令如下。[root@master1~]#start-all.sh執(zhí)行命令“jps”，查看當(dāng)前運(yùn)行的進(jìn)程，操作命令及結(jié)果如下。[root@master1~]#jps673ResourceManager503SecondaryNameNode935Jps298NameNode[root@slave1~]#jps256NodeManager388Jps140DataNode[root@slave2~]#jps243NodeManager375Jps127DataNode2．將本地文件上傳到HDFS中將本地文件test.txt上傳到HDFS中，并存儲在HDFS的/bigdatacase/dataset目錄中。在HDFS的根目錄中創(chuàng)建一個(gè)新的目錄bigdatacase，并在其中創(chuàng)建一個(gè)子目錄dataset，操作命令及結(jié)果如下。[root@master1~]#hadoopdfs-mkdir-p/bigdatacase/datasetDEPRECATED:Useofthisscripttoexecutehdfscommandisdeprecated.Insteadusethehdfscommandforit.執(zhí)行命令“hadoopdfs-put/opt/test.txt/bigdatacase/dataset”，將test.txt文件上傳到HDFS的/bigdatacase/dataset目錄中，操作命令及結(jié)果如下。[root@master1~]#hadoopdfs-put/opt/test.txt/bigdatacase/datasetDEPRECATED:Useofthisscripttoexecutehdfscommandisdeprecated.Insteadusethehdfscommandforit.

執(zhí)行命令“hadoopdfs-cat/bigdatacase/dataset/test.txt|head-10”，查看HDFS中的test.txt的前10條記錄，操作命令及結(jié)果如下。[root@master1~]#hadoopdfs-cat/bigdatacase/dataset/test.txt|head-10

DEPRECATED:Useofthisscripttoexecutehdfscommandisdeprecated.Insteadusethehdfscommandforit.43,2015/3/3017:38,/static/image/common/faq.gif26,2015/3/3017:38,/data/cache/style_1_widthauto.css?y7a43,2015/3/3017:38,/static/image/common/hot_1.gif43,2015/3/3017:38,/static/image/common/hot_2.gif43,2015/3/3017:38,/static/image/filetype/common.gif26,2015/3/3017:38,/source/plugin/wsh_wx/img/wsh_zk.css26,2015/3/3017:38,/data/cache/style_1_forum_index.css?y7a26,2015/3/3017:38,/source/plugin/wsh_wx/img/wx_jqr.gif43,2015/3/3017:38,/static/image/common/recommend_1.gif26,2015/3/3017:38,/static/image/common/logo.png3．在Hive中創(chuàng)建數(shù)據(jù)庫（1）創(chuàng)建數(shù)據(jù)庫和數(shù)據(jù)表執(zhí)行命令“servicemysqlstart”，啟動(dòng)MySQL數(shù)據(jù)庫，操作命令及結(jié)果如下。[root@master1~]#servicemysqlstartStartingMySQLSUCCESS!Hive是基于Hadoop的數(shù)據(jù)倉庫，使用HiveQL語言編寫的查詢語句，最終都會(huì)被Hive自動(dòng)解析為MapReduce任務(wù)，并由Hadoop具體執(zhí)行。因此，需要先啟動(dòng)Hadoop服務(wù)，再啟動(dòng)Hive服務(wù)，可通過執(zhí)行命令“./hive”來啟動(dòng)Hive服務(wù)，操作命令及結(jié)果如下。[root@master1bin]#./hive

Logginginitializedusingconfigurationinjar:file:/simple/hive1.2.1/lib/hive-common-1.2.1.jar!/pertieshive>啟動(dòng)Hive服務(wù)后，執(zhí)行命令“createdatabasedblab”，在Hive中創(chuàng)建一個(gè)數(shù)據(jù)庫dblab，操作命令及結(jié)果如下。hive>createdatabasedblab;OKTimetaken:0.592seconds創(chuàng)建外部表，操作命令及結(jié)果如下。hive>createexternaltabledblab.bigdata_user(ipstring,timestring,urlstring)rowformatdelimitedfieldsterminatedby','storedastextfilelocation'/bigdatacase/dataset';OKTimetaken:0.042seconds（2）查詢數(shù)據(jù)在Hive命令行模式下，執(zhí)行命令“showcreatetablebigdata_user”，查看表的各種屬性，操作命令及結(jié)果如下。（文本位于頁面下方）hive>showcreatetablebigdata_user;OKCREATEEXTERNALTABLE`bigdata_user`(`ip`string,`time`string,`url`string)ROWFORMATDELIMITEDFIELDSTERMINATEDBY','STOREDASINPUTFORMAT'org.apache.hadoop.mapred.TextInputFormat'OUTPUTFORMAT'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'LOCATION'hdfs://master1:9000/bigdatacase/dataset'TBLPROPERTIES('COLUMN_STATS_ACCURATE'='false','numFiles'='0','numRows'='-1','rawDataSize'='-1','totalSize'='0','transient_lastDdlTime'='1677831958')Timetaken:0.033seconds,Fetched:19row(s)執(zhí)行命令“descbigdata_user”，查看表的簡單結(jié)構(gòu)，操作命令及結(jié)果如下。hive>desc

bigdata_user;OKipstringtimestringurlstringTimetaken:0.049seconds,Fetched:3row(s)執(zhí)行命令“select*frombigdata_userlimit10”，查看表的前10條數(shù)據(jù)，操作命令及結(jié)果如下。hive>select*frombigdata_userlimit10;OK432015/3/3017:38/static/image/common/faq.gif262015/3/3017:38/data/cache/style_1_widthauto.css?y7a432015/3/3017:38/static/image/common/hot_1.gif432015/3/3017:38/static/image/common/hot_2.gif432015/3/3017:38/static/image/filetype/common.gif262015/3/3017:38/source/plugin/wsh_wx/img/wsh_zk.css262015/3/3017:38/data/cache/style_1_forum_index.css?y7a262015/3/3017:38/source/plugin/wsh_wx/img/wx_jqr.gif432015/3/3017:38/static/image/common/recommend_1.gif262015/3/3017:38/static/image/common/logo.pngTimetaken:0.046seconds,Fetched:10row(s)8.2

使用Hive進(jìn)行簡單的數(shù)據(jù)分析任務(wù)實(shí)施1．簡單查詢分析執(zhí)行命令“selectipfrombigdata_userlimit10”，查詢前10條記錄的ip，操作命令及結(jié)果如下。hive>selectipfrombigdata_userlimit10;OK43264343432626264326Timetaken:0.064seconds,Fetched:10row(s)2．查詢前20條記錄的ip和time執(zhí)行命令“selectip,timefrombigdata_userlimit20”，查詢前20條記錄的ip和time，操作命令及結(jié)果如下。3．使用聚合函數(shù)count()統(tǒng)計(jì)表中的數(shù)據(jù)執(zhí)行命令“selectcount(*)frombigdata_user”，統(tǒng)計(jì)表中的數(shù)據(jù)，操作命令及結(jié)果如下。8.3

Hive、MySQL、HBase數(shù)據(jù)的互導(dǎo)任務(wù)實(shí)施1．Hive預(yù)操作創(chuàng)建臨時(shí)表user_action，操作命令及結(jié)果如下。hive>createexternaltableuser_action(ipstring,timestring,urlstring)rowformatdelimitedfieldsterminatedby','storedastextfile;OKTimetaken:0.054seconds創(chuàng)建完成后，Hive會(huì)自動(dòng)在HDFS中創(chuàng)建對應(yīng)的數(shù)據(jù)文件/user/hive/warehouse/dbalb.db/user_action。執(zhí)行命令“hadoopdfs-ls/user/hive/warehouse/dblab.db/”，在HDFS中查看創(chuàng)建的user_action表，操作命令及結(jié)果如下。[root@master1/]#hadoopdfs-ls/user/hive/warehouse/dblab.db/

DEPRECATED:Useofthisscripttoexecutehdfscommandisdeprecated.Insteadusethehdfscommandforit.Found1itemsdrwxr-xr-x-rootsupergroup02023-03-0316:32/user/hive/warehouse/dblab.db/user_action2．?dāng)?shù)據(jù)導(dǎo)入操作在HiveShell模式下執(zhí)行命令“insertoverwritetabledblab.user_actionselect*fromdblab.bigdata_user”，將bigdata_user表中的數(shù)據(jù)導(dǎo)入到user_action表中，操作命令如下。hive>insertoverwritetabledblab.user_actionselect*fromdblab.bigdata_user;執(zhí)行命令“select*fromuser_actionlimit10”，查詢表的前10條記錄，操作命令及結(jié)果如下。3．使用Sqoop將數(shù)據(jù)從Hive導(dǎo)入到MySQL中登錄MySQL，在dblab數(shù)據(jù)庫中創(chuàng)建與Hive對應(yīng)的user_action表，并設(shè)置其編碼格式為UTF-8，操作命令及結(jié)果如下。mysql>usedblab;Databasechangedmysql>createtableuser_action(->ipvarchar(50),->timevarchar(50),->urlvarchar(255))->ENGINE=InnoDBDEFAULTCHARSET=utf8;QueryOK,0rowsaffected(0.00sec)退出MySQL，進(jìn)入Sqoop的bin目錄，導(dǎo)入數(shù)據(jù)，操作命令如下。[root@master1bin]#./sqoopexport--connectjdbc:mysql://master1:3306/dblab--usernameroot--password123456--tableuser_action--export-dir/user/hive/warehouse/dblab.db/user_action--input-fields-terminated-by',';使用root用戶登錄MySQL，查看已經(jīng)從Hive導(dǎo)入到MySQL中的數(shù)據(jù)，操作命令及結(jié)果如下。4．使用Sqoop將數(shù)據(jù)從MySQL導(dǎo)入到HBase中啟動(dòng)Hadoop集群和HBase服務(wù)，并查看集群節(jié)點(diǎn)進(jìn)程，操作命令及結(jié)果如下。[root@master1bin]#start-all.sh[root@master1bin]#./zkServer.shstart[root@master1bin]#./start-hbase.shmaster1節(jié)點(diǎn)的進(jìn)程如下。[root@master1bin]#jps1714SecondaryNameNode4437Jps3207HMaster1514NameNode1883ResourceManager3358HRegionServer3039QuorumPeerMainslave1節(jié)點(diǎn)的進(jìn)程如下。[root@slave1bin]#jps577NodeManager786QuorumPeerMain1811Jps854HRegionServer473DataNodeslave2節(jié)點(diǎn)的進(jìn)程如下。[root@slave2bin]#jps578NodeManager3154Jps1028QuorumPeerMain474DataNode1102HRegionServer進(jìn)入HBaseShell，操作命令及結(jié)果如下。在HBase中創(chuàng)建user_action表，操作命令及結(jié)果如下。hbase(main):001:0>create'user_action',{NAME=>'f1',VERSION=>5}0row(s)in1.4250seconds新建一個(gè)終端，導(dǎo)入數(shù)據(jù)，操作命令如下。[root@master1bin]#./sqoopimport--connectjdbc:mysql://localhost:3306/dblab?zeroDateTimeBehavior=ROUND--usernameroot--password123456--tableuser_action--hbase-tableuser_action--column-familyf1--hbase-row-keyip-m1再次切換到HBaseShell運(yùn)行的終端窗口，執(zhí)行命令“scan'user_action'”，查詢插入的數(shù)據(jù)，如圖所示。5．利用HBase-thrift庫將數(shù)據(jù)導(dǎo)入到HBase中首先，使用“pip”命令安裝最新版的HBase-thrift庫，操作命令如下。[root@master1bin]#pipinstallhbase-thrift其次，在hbase/bin目錄下啟動(dòng)thrift相關(guān)命令，開啟9095端口，操作命令如下。[root@master1bin]#hbase-daemon.shstartthrift--infoport9095-p9090查看9095端口，操作命令及結(jié)果如下所示。[root@master1~]#netstat-anp|grep9095tcp00:9095:*LISTEN1802/java執(zhí)行“jps”命令，查看進(jìn)程運(yùn)行情況，操作命令及結(jié)果如下。[root@master1~]#jps501SecondaryNameNode1125HMaster280NameNode1802ThriftServer1292HRegionServer942QuorumPeerMain1886Jps671ResourceManager在HBase中創(chuàng)建ip_info表，表中包含兩個(gè)列族分別為：ip、url，查看創(chuàng)建的表，操作命令及結(jié)果如下。hbase(main):004:0>create‘ip_info’,’ip’,’url’

=>["ip_info","user_action"]hbase(main):005:0>desc'ip_info'Tableip_infoisENABLEDip_infoCOLUMNFAMILIESDESCRIPTION{NAME=>'ip',BLOOMFILTER=>'ROW',VERSIONS=>'1',IN_MEMORY=>'false',KEEP_DELETED_CELLS=>'FALSE',DATA_BLOCK_ENCODING=>'NONE',TTL=>'FOREVER',COMPRESSION=>'NONE',MIN_VERSIONS=>'0',BLOCKCACHE=>'true',BLOCKSIZE=>'65536',REPLICATION_SCOPE=>'0'}{NAME=>'url',BLOOMFILTER=>'ROW',VERSIONS=>'1',IN_MEMORY=>'false',KEEP_DELETED_CELLS=>'FALSE',DATA_BLOCK_ENCODING=>'NONE',TTL=>'FOREVER',COMPRESSION=>'NONE',MIN_VERSIONS=>'0',BLOCKCACHE=>'true',BLOCKSIZE=>'65536',REPLICATION_SCOPE=>'0'}2row(s)in0.0820seconds

使用Python編程將本地?cái)?shù)據(jù)導(dǎo)入到HBase中，程序代碼如下。切換到HBaseShell運(yùn)行窗口，查詢ip_info中插入的3條數(shù)據(jù)，如圖所示。任務(wù)8.4

流數(shù)據(jù)處理的簡單應(yīng)用任務(wù)實(shí)施在進(jìn)行操作前，需確保Hadoop、Zookeeper和Kafka集群已啟動(dòng)。1.Telnet工具的安裝與使用(1)通過yum命令安裝Telnet和Netcat工具，操作命令如下。[root@master1~]#yuminstall-ytelnet[root@master1~]#yuminstall-ync(2)Netcat工具的使用測試，開啟一個(gè)本地7777的TCP協(xié)議端口，等待客戶端發(fā)起連接，操作命令及結(jié)果如下。[root@master1~]#nc-lk7777新打開一個(gè)終端窗口，執(zhí)行“telnet”命令，進(jìn)入telnet客戶端命令模式，操作命令及結(jié)果如下。[root@master1~]#telnet7777Trying...Connectedto.Escapecharacteris'^]'.在telnet客戶端命令模式下輸入如下信息，向目標(biāo)服務(wù)器發(fā)送“test”和“txt”信息，若要退出命令發(fā)送模式，直接按'^]'即可，然后執(zhí)行quit退出telnet客戶端，輸入信息如下。testtxt在“nc”監(jiān)聽窗口監(jiān)聽到打印數(shù)據(jù)代表測試成功，操作結(jié)果如下。[root@master1~]#nc-lk7777

testTxt如果Netcat順利監(jiān)聽到數(shù)據(jù)，即可終止上述兩個(gè)命令完成測試。2.Kafka主題的創(chuàng)建與查看(1)創(chuàng)建wordcount主題，分區(qū)數(shù)設(shè)為4，副本數(shù)為1，操作命令及結(jié)果如下。[root@master1bin]#./kafka-topics.sh--create--zookeepermaster1:2181,slave1:2181,slave2:2181--partitions4--replication-factor1--topicwordcountCreatedtopic"wordcount"(2)使用"--list"命令查看已經(jīng)創(chuàng)建好的主題列表，操作命令及結(jié)果如下。[root@master1bin]#./kafka-topics.sh--list--zookeepermaster1:2181,slave1:2181,slave2:2181__consumer_offsetstestwordcount(3)通過“--describe--topicwordcount”，可以查看wordcount主題的具體信息，操作命令及結(jié)果如下。[root@master1bin]#./kafka-topics.sh--describe--zookeepermaster1:2181,slave1:2181,slave2:2181--topicwordcountTopic:wordcount

PartitionCount:4ReplicationFactor:1Configs:Topic:wordcountPartition:0Leader:1Replicas:1Isr:1Topic:wordcountPartition:1Leader:2Replicas:2Isr:2Topic:wordcountPartition:2Leader:0Replicas:0Isr:0Topic:wordcountPartition:3Leader:1Replicas:1Isr:13.Flume配置與啟動(dòng)(1)切換到Flume的conf目錄下，操作命令如下。[root@master1~]#cd/usr/local/flume/conf/(2)新建并編輯一個(gè)配置文件nc_wordcount.conf，操作命令及配置文件如下。[root@master1conf]#vinc_wordcount.confa1.sources=s1a1.channels=c1a1.sources.s1.type=netcata1.sources.s1.bind=localhosta1.sources.s1.port=7777a1.channels.c1.type=org.apache.flume.channel.kafka.KafkaChannela1.channels.c1.kafka.bootstrap.servers=master1:9092,slave1:9092,slave2:9092a1.channels.c1.kafka.topic=wordcounta1.channels.c1.parseAsFlumeEvent=false

a1.sources.s1.channels=c1(3)啟動(dòng)Flume（操作窗口請不要關(guān)掉），并指定配置文件為nc_wordcount.conf，操作命令及部分結(jié)果如下。[root@master1flume]#flume-ngagent-na1-cconf--f./conf/nc_wordcount.conf-Dflume.root.logger=info,console.......2023-03-2714:13:58,868(lifecycleSupervisor-1-0)[INFO-org.apache.flume.instrumentation.MonitoredCounterGroup.start(MonitoredCounterGroup.java:95)]Componenttype:CHANNEL,name:c1started2023-03-2714:13:58,869(conf-file-poller-0)[INFO-org.apache.flume.node.Application.startAllComponents(Application.java:182)]StartingSources12023-03-2714:13:58,872(lifecycleSupervisor-1-1)[INFO-org.apache.flume.source.NetcatSource.start(NetcatSource.java:155)]Sourcestarting2023-03-2714:13:58,885(lifecycleSupervisor-1-1)[INFO-org.apache.flume.source.NetcatSource.start(NetcatSource.java:169)]CreatedserverSocket:sun.nio.ch.ServerSocketChannelImpl[/:7777]4.數(shù)據(jù)流寫入Kafka(1)新開一個(gè)終端窗口，通過“telnet7777”連接對應(yīng)端口，并向端口寫入一些單詞數(shù)據(jù)，操作命令及結(jié)果如下。[root@master1~]#telnet7777Trying...Connectedto.Escapecharacteris'^]'.helloworldOKhelloscalaOKhelloworldOKhelloworldOK(2)再次打開一個(gè)終端窗口，創(chuàng)建Kafka消費(fèi)者用于消費(fèi)“wordcount”主題，操作命令及結(jié)果如下。[root@master1bin]#./kafka-console-consumer.sh--bootstrap-servermaster1:9092,slave2:9092,slave2:9092--topicwordcount--from-beginninghelloworldhelloscalahelloworldhelloworld5.Flink編程（1）在IDEA中創(chuàng)建一個(gè)maven項(xiàng)目，命名為FlinkWordCount。需要在IDEA中安裝Scala插件，并勾選起用。（2）在項(xiàng)目的src/main下新建一個(gè)Directory，命名為scala，選中并右擊，選擇”MarkDirectoryas“，在二級菜單下選擇SourcesRoot。（3）在項(xiàng)目名上，右擊選擇OpenModuleSettings，在彈出的頁面中選擇Modules，單擊+號，選擇Scala，若本地沒有scala，則單擊Create...，再單擊Download...，選擇scala版本2.11.0,單擊OK。（4）編輯pom.xml文件，編輯文件內(nèi)容如下。

<?xmlversion="1.0"encoding="UTF-8"?><projectxmlns="/POM/4.0.0"

xmlns:xsi="/2001/XMLSchema-instance"

xsi:schemaLocation="/POM/4.0.0/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.example</groupId><artifactId>FlinkWordCount</artifactId><version>1.0-SNAPSHOT</version><properties><piler.source>8</piler.source><piler.target>8</piler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-scala_2.11</artifactId><version>1.10.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_2.11</artifactId><version>1.10.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka_2.11</artifactId><version>1.10.0</version></dependency></dependencies></project>（5）在scala文件夾下，新建scala文件，選擇Object型，命名為WordCount，編寫代碼如下。importorg.apache.flink.api.common.serialization.SimpleStringSchemaimportorg.apache.flink.streaming.api.TimeCharacteristicimportorg.apache.flink.streaming.api.scala._importorg.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer,

FlinkKafkaProducer}importorg.apache.kafka.clients.consumer.ConsumerConfigimportjava.util.PropertiesobjectWordCount{defmain(args:Array[String]):Unit={//創(chuàng)建流處理環(huán)境valenv=StreamExecutionEnvironment.getExecutionEnvironment

env.setParallelism(1)

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)//設(shè)置kafka連接屬性配置valproperties=newProperties()properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"master1:9092,slave1:9092,slave2:9092")//kafka集群properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest")//從主題的第一條消息開始讀取//鍵值反序列化配置properties.setProperty