Hadoop平臺(tái)搭建與應(yīng)用（第2版）（微課版）-教案 項(xiàng)目5 Hadoop生態(tài)組件的安裝與使用

第4頁Hadoop平臺(tái)搭建與應(yīng)用教案NO.5教師姓名授課班級(jí)授課形式理實(shí)一體化授課時(shí)間年月日第周授課時(shí)數(shù)16授課章節(jié)名稱項(xiàng)目5Hadoop生態(tài)組件的安裝與使用教學(xué)目的（1）識(shí)記Hadoop常用工具組件（Sqoop、Pig、Flume）的作用。（2）領(lǐng)會(huì)Hadoop各組件的功能與聯(lián)系。（4）學(xué)會(huì)Hadoop常用工具組件（Sqoop、Pig、Flume）的使用。教材分析教學(xué)重點(diǎn)教學(xué)難點(diǎn)Hadoop常用工具組件（Sqoop、Pig、Flume）的使用更新、補(bǔ)充、刪節(jié)內(nèi)容課外作業(yè)熟練Hadoop常用工具組件（Sqoop、Pig、Flume）的安裝教學(xué)過程教學(xué)提示項(xiàng)目

5Hadoop生態(tài)組件的安裝與使用任務(wù)5.1Sqoop的安裝與應(yīng)用Sqoop（SQL-to-Hadoop）是一個(gè)開源工具，主要用于在Hadoop（Hive）與傳統(tǒng)的數(shù)據(jù)庫（MySQL、Oracle等）之間進(jìn)行數(shù)據(jù)傳遞，可以將一個(gè)關(guān)系型數(shù)據(jù)庫（如MySQL、Oracle、PostgreSQL等）中的數(shù)據(jù)導(dǎo)入到Hadoop的HDFS中，也可以將HDFS的數(shù)據(jù)導(dǎo)入到關(guān)系型數(shù)據(jù)庫中。對于某些NoSQL數(shù)據(jù)庫，Sqoop也提供了連接器。Sqoop類似于其他ETL[Extract-Transform-Load，用來描述將數(shù)據(jù)從源端經(jīng)過萃取（Extract）、轉(zhuǎn)換（Transform）、加載（Load）至目的端的過程]工具，使用元數(shù)據(jù)模型來判斷數(shù)據(jù)類型，并在數(shù)據(jù)從數(shù)據(jù)源轉(zhuǎn)移到Hadoop時(shí)確保類型進(jìn)行安全的數(shù)據(jù)處理。Sqoop專為大數(shù)據(jù)批量傳輸設(shè)計(jì)，能夠通過分割數(shù)據(jù)集并創(chuàng)建Hadoop任務(wù)來處理每個(gè)區(qū)塊。Sqoop項(xiàng)目開始于2009年，最早作為Hadoop的一個(gè)第三方模塊存在，后來為了讓使用者快速部署，也為了使開發(fā)人員更快速地進(jìn)行迭代開發(fā)，Sqoop獨(dú)立成為一個(gè)Apache項(xiàng)目。1．Sqoop的核心功能Sqoop的核心功能包含以下2項(xiàng)。（1）導(dǎo)入數(shù)據(jù)：MySQL、Oracle導(dǎo)入數(shù)據(jù)到Hadoop的HDFS、Hive、HBase等數(shù)據(jù)存儲(chǔ)系統(tǒng)中。（2）導(dǎo)出數(shù)據(jù)：從Hadoop的文件系統(tǒng)中導(dǎo)出數(shù)據(jù)到關(guān)系型數(shù)據(jù)庫（MySQL等）中。Sqoop本質(zhì)上是一個(gè)命令行工具。2．Sqoop中import命令的使用（1）默認(rèn)情況下，通過使用“import”命令可導(dǎo)入數(shù)據(jù)到HDFS中。（2）指定目錄和Mapper個(gè)數(shù)，并導(dǎo)入到HDFS中。①創(chuàng)建目錄。${HADOOP_HOME}/bin/hdfsdfs-mkdir-p/user/sqoop/②設(shè)置Mapper個(gè)數(shù)為1，指定目錄為/bin/sqoop/，如果目標(biāo)目錄已經(jīng)存在，則先刪除原目錄，再創(chuàng)建新目錄。（3）將增量數(shù)據(jù)導(dǎo)入到HDFS文件中時(shí)，可以通過對“check-column”“incremental”“l(fā)ast-value”3個(gè)參數(shù)進(jìn)行設(shè)置來實(shí)現(xiàn)。（4）指定文件格式并導(dǎo)入HDFS。默認(rèn)情況下，導(dǎo)入數(shù)據(jù)到HDFS中時(shí)，文件存儲(chǔ)格式為textfile，可以通過對屬性進(jìn)行指定，以設(shè)定文件存儲(chǔ)格式為parquet。（5）指定壓縮格式并導(dǎo)入HDFS。默認(rèn)情況下，導(dǎo)入的HDFS文件是不壓縮的，可以通過屬性“compress”“compression-codec”進(jìn)行設(shè)置。（6）將Select查詢結(jié)果導(dǎo)入到HDFS中時(shí)，必須在Where子句中包含'$CONDITIONS'。導(dǎo)入數(shù)據(jù)到Hive中，實(shí)例如下。$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）這里以數(shù)據(jù)導(dǎo)入為例進(jìn)行說明。①導(dǎo)入數(shù)據(jù)到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"②執(zhí)行腳本。$bin/sqoop--options-file***.opt（2）從Hive或者HDFS中導(dǎo)出數(shù)據(jù)到MySQL中，實(shí)例如下。$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"任務(wù)5.2Pig的安裝和使用1．ApachePig概述ApachePig是MapReduce的一個(gè)抽象。它是一個(gè)工具/平臺(tái)，用于分析較大的數(shù)據(jù)集，并將數(shù)據(jù)集表示為數(shù)據(jù)流。Pig通常與Hadoop一起使用，可以使用ApachePig在Hadoop中執(zhí)行所有的數(shù)據(jù)處理操作。當(dāng)要編寫數(shù)據(jù)分析程序時(shí)，Pig中提供了一種稱為PigLatin的高級(jí)語言。該語言提供了各種操作符，程序員可以利用其開發(fā)自己的用于讀取、寫入和處理數(shù)據(jù)的程序。要想使用ApachePig分析數(shù)據(jù)，程序員需要使用PigLatin語言編寫腳本。所有腳本都在內(nèi)部轉(zhuǎn)換為Map和Reduce任務(wù)。Pig的工作原理如圖5-3所示。圖5-3Pig的工作原理2．PigLatin的數(shù)據(jù)模型PigLatin的數(shù)據(jù)模型是完全嵌套的，它允許使用復(fù)雜的非原子數(shù)據(jù)類型，如Map和Tuple。PigLatin的數(shù)據(jù)模型。（1）Atom（原子）：PigLatin中的任何數(shù)據(jù)類型的單個(gè)值都稱為Atom。它存儲(chǔ)為字符串，可以用作字符串和數(shù)字。Int、long、float、double、chararray和bytearray都是Pig的原子值。一條數(shù)據(jù)或一個(gè)簡單的原子值被稱為字段，如“raja”或“30”。（2）Tuple（元組）：由有序字段集合形成的記錄稱為元組，字段可以是任意類型。元組與關(guān)系數(shù)據(jù)庫管理系統(tǒng)表中的行類似，如（Raja，30）。（3）Bag（包）：一個(gè)包是一組無序的元組。每個(gè)元組可以有任意數(shù)量的字段（靈活模式）。包由“{}”表示。它類似于關(guān)系數(shù)據(jù)庫管理系統(tǒng)中的表，但是與關(guān)系數(shù)據(jù)庫管理系統(tǒng)中的表不同的是，包不需要每個(gè)元組中都包含相同數(shù)量的字段，或者相同位置（列）中的字段具有相同類型。例如，{（Raja，30），（Mohammad，45）}。（4）Map（映射）：映射（或數(shù)據(jù)映射）是一組Key/Value對。其中，Key必須是字符數(shù)組類型，且應(yīng)該是唯一的；Value可以是任何類型，它由“[]”表示。例如，['name''Raja'，'age'30]。（5）Relation（關(guān)系）：一個(gè)關(guān)系是一個(gè)元組的包。PigLatin中的關(guān)系是無序的（不能保證按任何特定順序處理元組）。3．Pig語句基礎(chǔ)在使用PigLatin處理數(shù)據(jù)時(shí)，語句是基本結(jié)構(gòu)。每個(gè)語句以分號(hào)（;）結(jié)尾。使用PigLatin提供的運(yùn)算符可通過語句執(zhí)行各種操作。除了LOAD和STORE語句之外，在執(zhí)行其他操作時(shí)，PigLatin語句采用關(guān)系作為輸入，并產(chǎn)生另一個(gè)關(guān)系作為輸出。只要在Shell中輸入Load語句，就會(huì)執(zhí)行語義檢查。要查看模式的內(nèi)容，需要使用Dump運(yùn)算符。只有在執(zhí)行Dump操作后，才會(huì)執(zhí)行將數(shù)據(jù)加載到文件系統(tǒng)中的MapReduce任務(wù)。PigLatin的數(shù)據(jù)類型如表5-1所示。表5-1PigLatin的數(shù)據(jù)類型序號(hào)數(shù)據(jù)類型說明＆示例1int表示有符號(hào)的32位整數(shù)。示例：82long表示有符號(hào)的64位整數(shù)。示例：5L3float表示有符號(hào)的32位浮點(diǎn)數(shù)。示例：5.5F4double表示有符號(hào)的64位浮點(diǎn)數(shù)。示例：10.55chararray表示UnicodeUTF-8格式的字符數(shù)組。示例：'w3cschool'6bytearray表示字節(jié)數(shù)組7boolean表示布爾值。示例：true/false8datetime表示日期時(shí)間。示例：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]上述數(shù)據(jù)類型的值可以為null。ApachePig以與SQL類似的方式處理空值。null可以是未知值或不存在值，它用作可選值的占位符。PigLatin的結(jié)構(gòu)運(yùn)算符如表5-2所示。表5-2PigLatin的結(jié)構(gòu)運(yùn)算符運(yùn)算符描述示例()元組構(gòu)造函數(shù)運(yùn)算符，此運(yùn)算符用于構(gòu)建元組（Raju,30）{}包構(gòu)造函數(shù)運(yùn)算符，此運(yùn)算符用于構(gòu)造包{（Raju,30）,（Mohammad,45）}[]映射構(gòu)造函數(shù)運(yùn)算符，此運(yùn)算符用于構(gòu)造一個(gè)映射[nameRaja,age30]PigLatin的關(guān)系運(yùn)算符如表5-3所示。表5-3PigLatin的關(guān)系運(yùn)算符運(yùn)算符描述加載和存儲(chǔ)LOAD將數(shù)據(jù)從文件系統(tǒng)（local/HDFS）加載到關(guān)系中STORE將數(shù)據(jù)從文件系統(tǒng)（local/HDFS）存儲(chǔ)到關(guān)系中過濾FILTER從關(guān)系中刪除不需要的行DISTINCT從關(guān)系中刪除重復(fù)行FOREACH，GENERATE基于數(shù)據(jù)列生成數(shù)據(jù)轉(zhuǎn)換STREAM使用外部程序轉(zhuǎn)換關(guān)系分組和連接JOIN連接兩個(gè)或多個(gè)關(guān)系COGROUP將數(shù)據(jù)分組為兩個(gè)或多個(gè)關(guān)系GROUP在單個(gè)關(guān)系中對數(shù)據(jù)進(jìn)行分組CROSS創(chuàng)建兩個(gè)或多個(gè)關(guān)系的向量積排序ORDER基于一個(gè)或多個(gè)字段按排序（升序或降序）排列關(guān)系LIMIT從關(guān)系中獲取有限數(shù)量的元組合并和拆分UNION將兩個(gè)或多個(gè)關(guān)系合并為單個(gè)關(guān)系SPLIT將單個(gè)關(guān)系拆分為兩個(gè)或多個(gè)關(guān)系診斷運(yùn)算符DUMP在控制臺(tái)中輸出關(guān)系的內(nèi)容DESCRIBE描述關(guān)系的模式EXPLAIN查看邏輯、物理或MapReduce執(zhí)行計(jì)劃以計(jì)算關(guān)系ILLUSTRATE查看一行4．輸入和輸出（1）加載任何一種數(shù)據(jù)流的第一步都是指定輸入。在PigLatin中，通過LOAD語句來完成輸入操作。默認(rèn)情況下，LOAD使用默認(rèn)加載函數(shù)PigStorage，加載存放在HDFS中且以“Tab”鍵進(jìn)行分割的文件，例如，divs=load'pig_test'。用戶也可以通過指定一個(gè)完整的URL路徑來加載文件，如hdfs:///data/pig_test，其表示可以從NameNode為的HDFS中讀取文件。實(shí)際上，用戶的大部分?jǐn)?shù)據(jù)并非是使用“Tab”鍵作為分割符的文本文件，也有可能需要從其他非HDFS的存儲(chǔ)系統(tǒng)中加載數(shù)據(jù)。Pig允許用戶在加載數(shù)據(jù)時(shí)通過using句式指定其他加載函數(shù)。例如，從HBase中加載數(shù)據(jù)，語句如下。divs=load'pig_test'usingHBasestorage();如果沒有指定加載函數(shù)，那么會(huì)使用內(nèi)置的加載函數(shù)PigStorage。用戶同樣可以通過using句式為使用的加載函數(shù)指定參數(shù)。例如，如果想讀取以逗號(hào)分割的文本文件數(shù)據(jù)，那么PigStorage會(huì)接收一個(gè)指定分割符的參數(shù)，語句如下。divs=load'pig_test'usingPigStorage(',');LOAD語句也可以有as句式，這個(gè)句式可以為用戶加載的數(shù)據(jù)指定模式，模式相當(dāng)于列名。當(dāng)從HDFS中訪問指定“文件”的時(shí)候，用戶也可以指定目錄。在這種情況下，Pig會(huì)遍歷用戶指定的目錄中的所有文件，并將它們作為LOAD語句的輸入。PigStorage和TextLoader是內(nèi)置的可操作HDFS文件的Pig加載函數(shù)，是支持模式匹配的。通過模式匹配，用戶可以讀取不在同一個(gè)目錄中的多個(gè)文件，或者讀取一個(gè)目錄中的部分文件。其正則匹配語法如下：？匹配任意單個(gè)字符；*匹配零個(gè)或多個(gè)字符；[abc]匹配字符集合{a,b,c}所包含的任何一個(gè)字符，[a-z]匹配指定范圍內(nèi)的任意字符；[^abc]匹配未包含的任意字符，其中^符號(hào)匹配輸入字符串的開始位置；[^a-z]匹配不在指定范圍內(nèi)的任意字符；\c移除（轉(zhuǎn)義）字符c所表達(dá)的特殊含義；{ab,cd}匹配字符串集合{ab,cd}中的任一字符串。（2）存儲(chǔ)當(dāng)用戶處理完數(shù)據(jù)之后，需要把結(jié)果寫到某個(gè)地方。Pig提供了STORE語句進(jìn)行寫入數(shù)據(jù)的操作。默認(rèn)情況下，Pig使用PigStorage將結(jié)果數(shù)據(jù)以“Tab”鍵作為分割符，存儲(chǔ)到HDFS的“storeprocessedinto'/data'”中。如果用戶沒有顯式指定存儲(chǔ)函數(shù)，那么將會(huì)默認(rèn)使用PigStorage。用戶可以使用using句式指定不同的存儲(chǔ)函數(shù)。例如：storeprocessedinto'/data'usingHBaseStorage();用戶也可以傳參數(shù)給其使用的存儲(chǔ)函數(shù)。例如，如果想將數(shù)據(jù)存儲(chǔ)為以逗號(hào)分隔的文本數(shù)據(jù)，則PigStorage會(huì)接收一個(gè)指定分隔符的參數(shù)“storeprocessedinto'/data'usingPigStorage(',');”。當(dāng)寫到文件系統(tǒng)中后，data目錄中包含多個(gè)文件，而不是一個(gè)文件。但是到底會(huì)生成多少個(gè)文件要取決于執(zhí)行STORE操作前的最后一個(gè)任務(wù)的并行數(shù)（該值由為這個(gè)任務(wù)所設(shè)置的并行級(jí)別所決定）。（3）輸出可以將關(guān)于結(jié)果的數(shù)據(jù)輸出到屏幕上，這在調(diào)試階段和原型研究階段是特別有用的。dump語句可以將用戶的腳本輸出到屏幕上（即“dumpprocessed”）。5．PigLatin常用操作（1）查詢固定行數(shù)據(jù)tmp_table_limit=limittmp_table50;dumptmp_table_limit;（2）查詢指定列數(shù)據(jù)tmp_table_name=foreachtmp_tablegeneratename;dumptmp_table_name;（3）為列取別名tmp_table_column_alias=foreachtmp_tablegeneratenameasusername,ageasuserage;dumptmp_table_column_alias;（4）按某列進(jìn)行排序tmp_table_order=ordertmp_tablebyageasc;dumptmp_table_order;（5）按條件進(jìn)行查詢tmp_table_where=filtertmp_tablebyage>18;dumptmp_table_where;（6）內(nèi)連接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）分組并統(tǒng)計(jì)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;任務(wù)5.3Flume的安裝和使用Flume作為Cloudera開發(fā)的實(shí)時(shí)日志收集系統(tǒng)，受到了業(yè)界的認(rèn)可與廣泛應(yīng)用。Flume初始的發(fā)行版本目前被統(tǒng)稱為FlumeOG（OriginalGeneration），屬于Cloudera。但隨著Flume功能的擴(kuò)展，F(xiàn)lumeOG代碼工程臃腫、核心組件設(shè)計(jì)不合理、核心配置不標(biāo)準(zhǔn)等缺點(diǎn)暴露出來，尤其是在FlumeOG的最后一個(gè)發(fā)行版本0.9.4中，日志傳輸不穩(wěn)定的現(xiàn)象尤為嚴(yán)重，為了解決這些問題，2011年10月22日，Cloudera完成了Flume-728的開發(fā)，對Flume進(jìn)行了里程碑式的改動(dòng)，重構(gòu)了其核心組件、核心配置及代碼架構(gòu)，重構(gòu)后的版本統(tǒng)稱為FlumeNG（NextGeneration）；這一改動(dòng)的另一原因是需將Flume納入Apache旗下，ClouderaFlume改名為ApacheFlume。Flume是一個(gè)分布式的、可靠的和高可用的海量日志采集、聚合和傳輸系統(tǒng)。其支持在日志系統(tǒng)中定制各類數(shù)據(jù)發(fā)送方，用于收集數(shù)據(jù)。同時(shí)，F(xiàn)lume提供對數(shù)據(jù)進(jìn)行簡單處理，并寫到各種數(shù)據(jù)接收方（如文本、HDFS、HBase等）的功能。Flume的數(shù)據(jù)流由事件(Event)貫穿始終。事件是Flume的基本數(shù)據(jù)單位，它攜帶有日志數(shù)據(jù)(字節(jié)數(shù)組形式)及頭信息，這些Event由Agent外部的Source生成，當(dāng)Source捕獲事件后，會(huì)進(jìn)行特定的格式化，且Source會(huì)把事件推入到(單個(gè)或多個(gè))Channel中?？梢园袰hannel看作一個(gè)緩沖區(qū)，它將保存事件直到Sink處理完該事件。Sink負(fù)責(zé)持久化日志或者把事件推向另一個(gè)Source。Flume主要由以下3個(gè)重要的組件構(gòu)成。（1）Source：完成對日志數(shù)據(jù)的收集，分為Transtion和Event并推入到Channel之中，F(xiàn)lume提供了各種Source的實(shí)現(xiàn)，包括AvroSource、ExceSource、SpoolingDirectorySource、NetCatSource、SyslogSource、SyslogTCPSource、SyslogUDPSource、HTTPSource、HDFSSource等。（2）Channel：主要提供隊(duì)列的功能，對Source提供的數(shù)據(jù)進(jìn)行簡單的緩存。Flume中的Channel有MemoryChannel、JDBCChanel、FileChannel等。（3）Sink：用于取出Channel中的數(shù)據(jù)，并將其存儲(chǔ)到文件系統(tǒng)、數(shù)據(jù)庫或者提交到遠(yuǎn)程服務(wù)器中。其包括HDFSSink、LoggerSink、AvroSink、FileRollSink、NullSink、HBaseSink等。任務(wù)5.4Kafka的安裝和使用Kafka是一個(gè)分布式、支持分區(qū)的（partition）、多副本的（replica），基于zookeeper協(xié)調(diào)的分布式消息系統(tǒng)，由服務(wù)器和客戶端組成，通過高性能TCP網(wǎng)絡(luò)協(xié)議進(jìn)行通信。它可以部署在本地和云環(huán)境中的裸機(jī)硬件、虛擬機(jī)和容器上。服務(wù)器：Kafka作為一個(gè)或多個(gè)服務(wù)器的集群運(yùn)行，這些服務(wù)器可以跨越多個(gè)數(shù)據(jù)中心或云區(qū)域。其中一些服務(wù)器形成了存儲(chǔ)層，稱為代理。其他服務(wù)器運(yùn)行KafkaConnect，將數(shù)據(jù)作為事件流不斷導(dǎo)入和導(dǎo)出，以將Kafka與現(xiàn)有系統(tǒng)（如關(guān)系數(shù)據(jù)庫和其他Kafka集群）集成。客戶端：它們允許編寫分布式應(yīng)用程序和微服務(wù)，即使在網(wǎng)絡(luò)問題或機(jī)器故障的情況下，也可以并行、大規(guī)模、容錯(cuò)地讀取、寫入和處理事件流。Kafka是一個(gè)事件流平臺(tái)，其以事件流的形式從數(shù)據(jù)庫、傳感器、移動(dòng)設(shè)備、云服務(wù)和軟件應(yīng)用程序等事件源實(shí)時(shí)捕獲數(shù)據(jù)，持久地存儲(chǔ)這些事件流以供以后檢索，實(shí)時(shí)地以及回顧性地操縱、處理和響應(yīng)事件流，或根據(jù)需要將事件流路由到不同的目的地，Kafka最大的特性就是可以實(shí)時(shí)的處理大量數(shù)據(jù)以滿足各種需求場景。Kafka可以與Flume、SparkStreaming、Storm、HBase、Flink和Spark協(xié)同工作，對流數(shù)據(jù)進(jìn)行實(shí)時(shí)讀取、分析和處理,應(yīng)用實(shí)例如下。（1）實(shí)時(shí)處理支付和金融交易，例如在證券交易所、銀行和保險(xiǎn)中。（2）實(shí)時(shí)跟蹤和監(jiān)控汽車、卡車、車隊(duì)和貨運(yùn)，如物流和汽車行業(yè)。（3）持續(xù)捕獲和分析來自物聯(lián)網(wǎng)設(shè)備或其他設(shè)備（如工廠和風(fēng)電場）的傳感器數(shù)據(jù)。（4）收集客戶互動(dòng)和訂單并立即做出反應(yīng)，例如在零售、酒店和旅游業(yè)以及移動(dòng)應(yīng)用程序中。（5）監(jiān)測醫(yī)院護(hù)理中的患者并預(yù)測病情變化，以確保在緊急情況下及時(shí)治療。（6）連接、存儲(chǔ)并提供由公司不同部門生成的數(shù)據(jù)。（7）作為數(shù)據(jù)平臺(tái)、事件驅(qū)動(dòng)架構(gòu)和微服務(wù)的基礎(chǔ)。一個(gè)事件記錄了“發(fā)生的一些事情”的事實(shí)，也稱為記錄或消息。當(dāng)讀或?qū)憯?shù)據(jù)到Kafka時(shí)，是以事件的形式來完成的。從概念上講，事件有一個(gè)鍵（Key）、值（Value）、時(shí)間戳（Timestamp）和可選的元數(shù)據(jù)（Metadata）頭。事件示例如下。事件的Key：“佩奇”；事件的Value：“向喬治支付了100元”；事件的Timestamp：“2023年1月1日下午1:11”。生產(chǎn)者是那些向Kafka發(fā)布（寫入）事件的客戶端應(yīng)用程序，消費(fèi)者是那些訂閱（讀取和處理）這些事件的人、應(yīng)用程序等。在Kafka中，生產(chǎn)者和消費(fèi)者是完全解耦的，彼此不可知，這是實(shí)現(xiàn)Kafka高可擴(kuò)展性的關(guān)鍵。例如，生產(chǎn)商永遠(yuǎn)不需要等待消費(fèi)者。事件按主題進(jìn)行組織和持久存儲(chǔ)。主題類似于文件系統(tǒng)中的目錄，事件是該目錄中的文件。例如主題名稱可以是“支付”。Kafka中的主題總是多生產(chǎn)者和多訂閱者，即一個(gè)主題可以有零個(gè)、一個(gè)或多個(gè)向其寫入事件的生產(chǎn)者，也可以有零、一個(gè)、或多個(gè)訂閱這些事件的消費(fèi)者?？梢愿鶕?jù)需要隨時(shí)讀取主題中的事件。與傳統(tǒng)的消息傳遞系統(tǒng)不同，使用后不會(huì)刪除事件。相反，可以通過每個(gè)主題的配置設(shè)置來定義Kafka應(yīng)該保留事件多長時(shí)間，之后舊事件將被丟棄。主題是分區(qū)的，這意味著一個(gè)主題分布在位于不同Kafka代理上的多個(gè)“桶”上。這種數(shù)據(jù)的分布式放置對于可擴(kuò)展性非常重要，因?yàn)樗试S客戶端應(yīng)用程序同時(shí)從多個(gè)代理讀取數(shù)據(jù)和向多個(gè)代理寫入數(shù)據(jù)。當(dāng)一個(gè)新事件發(fā)布到一個(gè)主題時(shí)，它實(shí)際上被附加到該主題的一個(gè)分區(qū)。具有相同Key（例如，客戶或車輛ID，在圖中用它們的顏色表示）的事件被寫入同一分區(qū)，Kafka保證給定主題分區(qū)的任何消費(fèi)者將始終以與寫入時(shí)完全相同的順序讀取該分區(qū)的事件。為了使數(shù)據(jù)具有容錯(cuò)性和高可用性，每個(gè)主題都可以復(fù)制，甚至可以跨地理區(qū)域或數(shù)據(jù)中心復(fù)制，這樣總是有多個(gè)代理擁有一份數(shù)據(jù)副本，以防出現(xiàn)問題，默認(rèn)副本參數(shù)設(shè)置為3，即始終有三份數(shù)據(jù)副本，該復(fù)制是在主題分區(qū)級(jí)別執(zhí)行的。除了用于管理和管理任務(wù)的命令行工具外，Kafka還有五個(gè)用于Java和Scala的核心API：AdminAPI：用于管理和查看主題、代理和其他Kafka對象。ProducerAPI：用于向一個(gè)或多個(gè)Kafka主題發(fā)布（編寫）事件流。ConsumerAPI：用于訂閱（讀?。┮粋€(gè)或多個(gè)主題并處理為其生成的事件流。KafkaStreamsAPI：用于實(shí)現(xiàn)流處理應(yīng)用程序和微服務(wù)。它提供了處理事件流更高級(jí)別的功能，包括轉(zhuǎn)換、聚合、聯(lián)接等。從一個(gè)或多個(gè)主題讀取輸入，以便生成一個(gè)或更多主題的輸出，從而有效地將輸入流轉(zhuǎn)換為輸出流。KafkaConnectAPI用于構(gòu)建和運(yùn)行可重復(fù)使用的數(shù)據(jù)導(dǎo)入/導(dǎo)出連接器，這些連接器從外部系統(tǒng)和應(yīng)用程序消費(fèi)（讀取）或生產(chǎn)（寫入）事件流，以便與Kafka集成。在實(shí)踐中通常不需要實(shí)現(xiàn)自己的連接器，因?yàn)镵afka社區(qū)已經(jīng)提供了數(shù)百個(gè)現(xiàn)成的連接器。任務(wù)5.5Flink的安裝和使用ApacheFlink是一個(gè)面向數(shù)據(jù)流處理和批量數(shù)據(jù)處理的分布式開源計(jì)算框架，它基于Flink流式執(zhí)行模型（streamingexecutionmodel），能夠支持流處理和批處理兩種應(yīng)用類型,適用于各種實(shí)時(shí)數(shù)據(jù)處理場景，可以對無界和有界數(shù)據(jù)流（有界數(shù)據(jù)流指的是有限大小的數(shù)據(jù)集，而無界數(shù)據(jù)流指的是無限的數(shù)據(jù)流，例如來自傳感器、日志、消息隊(duì)列等源的數(shù)據(jù)。）進(jìn)行高效、準(zhǔn)確、可靠的實(shí)時(shí)計(jì)算和數(shù)據(jù)處理。Flink的架構(gòu)如圖5-9所示，F(xiàn)link部署(Deploy)，支持local，和集群(standalone，yarn)，也支持在云上部署。Flink的核心（core），分布式的流處理引擎。Flink的API和類庫，有DataStreamAPI和DataSetAPI，分別做流處理和批處理。DataStreamAPI支持復(fù)雜事件處理，table操作和SQL操作。DatasetAPI支持FlinkML機(jī)器學(xué)習(xí)，Gelly圖計(jì)算，table操作和SQL操作。Flink包含三大核心組件.DataSource，數(shù)據(jù)源(負(fù)責(zé)接收數(shù)據(jù))；Transformations算子(負(fù)責(zé)對數(shù)據(jù)進(jìn)行處理)；DataSink輸出組件(負(fù)責(zé)把計(jì)算好的數(shù)據(jù)輸出到其它存儲(chǔ)介質(zhì)中)。Flink系統(tǒng)的架構(gòu)是基于Master-Slave架構(gòu)。Flink集群啟動(dòng)時(shí)，會(huì)啟動(dòng)一個(gè)JobManager進(jìn)程、至少一個(gè)TaskManager進(jìn)程。在Local模式下，會(huì)在同一個(gè)JVM內(nèi)部啟動(dòng)一個(gè)JobManager進(jìn)程和TaskManager進(jìn)程。JobManager負(fù)責(zé)接受并管理作業(yè)（Job），TaskManager負(fù)責(zé)執(zhí)行任務(wù)（Task）。一個(gè)作業(yè)由一個(gè)或多個(gè)任務(wù)組成，這些任務(wù)可以分配到不同的TaskManager上運(yùn)行。Ta