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

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

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