《大數(shù)據(jù)技術(shù)實(shí)戰(zhàn)案例教程》課件教學(xué)課件-第7章Kafka實(shí)戰(zhàn)

【第6章回顧】1.了解Flume功能、來(lái)源、特點(diǎn)和版本。2.理解Flume體系架構(gòu)及Source、Sink、Channel功能。（重點(diǎn)）3.理解Flume部署要點(diǎn)包括運(yùn)行環(huán)境、運(yùn)行模式、配置文件flume-env.sh。（重點(diǎn)）4.掌握FlumeShell命令的使用。（重點(diǎn)，難點(diǎn)）5.熟練掌握在Linux環(huán)境下部署Flume，靈活編寫Agent屬性文件和使用FlumeShell命令進(jìn)行實(shí)時(shí)日志收集。（重點(diǎn)，難點(diǎn)）【課程內(nèi)容】開學(xué)第一課（理論1學(xué)時(shí)）第1章部署全分布模式Hadoop集群（理論3學(xué)時(shí)+實(shí)驗(yàn)2學(xué)時(shí)）第2章HDFS實(shí)戰(zhàn)（理論4學(xué)時(shí)+實(shí)驗(yàn)2學(xué)時(shí)）第3章MapReduce編程（理論4學(xué)時(shí)+實(shí)驗(yàn)2學(xué)時(shí)）第4章部署ZooKeeper集群和ZooKeeper實(shí)戰(zhàn)（理論4學(xué)時(shí)+實(shí)驗(yàn)2學(xué)時(shí)）第5章部署本地模式Hive和Hive實(shí)戰(zhàn)（理論4學(xué)時(shí)+實(shí)驗(yàn)2學(xué)時(shí)）第6章Flume實(shí)戰(zhàn)（理論4學(xué)時(shí)+實(shí)驗(yàn)2學(xué)時(shí)）第7章Kafka實(shí)戰(zhàn)（理論4學(xué)時(shí)+實(shí)驗(yàn)2學(xué)時(shí)）第8章Spark集群部署和基本編程（理論4學(xué)時(shí)+實(shí)驗(yàn)2學(xué)時(shí)）第9章Flink集群部署和基本編程（選修）第7章Kafka實(shí)戰(zhàn)7.1初識(shí)Kafka7.2Kafka體系架構(gòu)7.3Kafka部署要點(diǎn)7.4KafkaShell常用命令7.5綜合實(shí)戰(zhàn)：Kafka實(shí)戰(zhàn)7.1初識(shí)KafkaApacheKafka是一個(gè)分布式的、支持分區(qū)的、多副本的、基于ZooKeeper的發(fā)布/訂閱消息系統(tǒng)，起源于LinkedIn公司開源出來(lái)的分布式消息系統(tǒng)，2011年成為Apache開源項(xiàng)目，2012年成為Apache頂級(jí)項(xiàng)目，目前被多家公司采用。Kafka采用Scala和Java編寫，其設(shè)計(jì)目的是通過Hadoop和Spark等并行加載機(jī)制來(lái)統(tǒng)一在線和離線的消息處理，構(gòu)建在ZooKeeper上，不同的分布式系統(tǒng)可統(tǒng)一接入到Kafka，實(shí)現(xiàn)和Hadoop各組件之間不同數(shù)據(jù)的實(shí)時(shí)高效交換，被稱為“生態(tài)系統(tǒng)的交通樞紐”。目前與越來(lái)越多的分布式處理系統(tǒng)如ApacheStorm、ApacheSpark等都能夠較好的集成，用于實(shí)時(shí)流式數(shù)據(jù)分析。7.2Kafka體系架構(gòu)訂閱消息發(fā)布消息到Partition

ConsumerGroup2

ConsumerGroup1Producer1Producer2Producer3Producer4KafkaBroker1KafkaBroker2KafkaBroker3Consumer1Consumer2Consumer3Consumer4Consumer5ZooKeeper7.2Kafka體系架構(gòu)Kafka整體架構(gòu)比較新穎，更適合異構(gòu)集群。Kafka中主要有Producer、Broker和Customer三種角色，一個(gè)典型的Kafka集群包含多個(gè)Producer、多個(gè)Broker、多個(gè)ConsumerGroup和一個(gè)ZooKeeper集群。每個(gè)Producer可以對(duì)應(yīng)多個(gè)Topic，每個(gè)Consumer只能對(duì)應(yīng)一個(gè)ConsumerGroup，整個(gè)Kafka集群對(duì)應(yīng)一個(gè)ZooKeeper集群，通過ZooKeeper管理集群配置、選舉Leader以及在ConsumerGroup發(fā)生變化時(shí)進(jìn)行負(fù)載均衡。7.2Kafka體系架構(gòu)（1）Message（消息）：Message是通信的基本單位，每個(gè)Producer可以向一個(gè)Topic發(fā)布一些消息，Kafka中的消息是以Topic為基本單位組織的，消息是無(wú)狀態(tài)的，消息消費(fèi)的先后順序是沒有關(guān)系的。每條Message包含三個(gè)屬性：offset，消息的唯一標(biāo)識(shí)，類型為long；MessageSize，消息的大小，類型為int；data，消息的具體內(nèi)容，可以看作一個(gè)字節(jié)數(shù)組。（2）Topic（主題）發(fā)布到Kafka集群的消息都有一個(gè)類別，這個(gè)類別被稱為Topic，Kafka根據(jù)Topic對(duì)消息進(jìn)行歸類，發(fā)布到Kafka集群的每條消息都需要指定一個(gè)Topic。（3）Partition（分區(qū)）：物理上的概念，一個(gè)Topic可以分為多個(gè)Partition，每個(gè)Partition內(nèi)部都是有序的。每個(gè)Partition只能由一個(gè)Consumer來(lái)進(jìn)行消費(fèi)，但是一個(gè)Consumer可以消費(fèi)多個(gè)Partition。（4）Broker：消息中間件處理節(jié)點(diǎn)。一個(gè)Kafka集群由多個(gè)Kafka實(shí)例組成，每個(gè)實(shí)例被稱為Broker。一個(gè)Broker上可以創(chuàng)建一個(gè)或多個(gè)Topic，同一個(gè)Topic可以在同一Kafka集群下的多個(gè)Broker上分布。Broker與Topic關(guān)系圖

Broker1Topic1Topic2Topic3Topic4

Broker2Topic1Topic2Topic3

Broker3Topic1Topic27.2Kafka體系架構(gòu)（5）Producer（消息生產(chǎn)者）：向Broker發(fā)送消息的客戶端。（6）Consumer（消息消費(fèi)者）：從Broker讀取消息的客戶端。（7）ConsumerGroup：每個(gè)Consumer屬于一個(gè)特定的ConsumerGroup，一條消息可以發(fā)送到多個(gè)不同的ConsumerGroup，但是一個(gè)ConsumerGroup中只能有一個(gè)Consumer能夠消費(fèi)該消息。7.3Kafka部署要點(diǎn)7.3.1Kafka運(yùn)行環(huán)境7.3.2Kafka運(yùn)行模式7.3.3Kafka配置文件7.3.1Kafka運(yùn)行環(huán)境1）操作系統(tǒng)Kafka支持不同操作系統(tǒng)，例如GNU/Linux、Windows、MacOSX等。需要注意的是，在Linux上部署Kafka要比在Windows上部署能夠得到更高效的I/O處理性能。編者采用的操作系統(tǒng)為L(zhǎng)inux發(fā)行版CentOS7。2）Java環(huán)境Kafka使用Java語(yǔ)言編寫，因此它的運(yùn)行環(huán)境需要Java環(huán)境的支持。編者采用的Java為OracleJDK1.8。3）ZooKeeper集群Kafka依賴ZooKeeper集群，因此運(yùn)行Kafka之前需要首先啟動(dòng)ZooKeeper集群。Zookeeper集群可以自己搭建，也可以使用Kafka安裝包中內(nèi)置的shell腳本啟動(dòng)Zookeeper。編者采用自行搭建ZooKeeper集群，版本為3.4.13。7.3.2Kafka運(yùn)行模式Kafka有兩種運(yùn)行模式：?jiǎn)螜C(jī)模式和集群模式。單機(jī)模式是只在一臺(tái)機(jī)器上安裝Kafka，主要用于開發(fā)測(cè)試，而集群模式則是在多臺(tái)機(jī)器上安裝Kafka，也可以在一臺(tái)機(jī)器上模擬集群模式，實(shí)際的生產(chǎn)環(huán)境中均采用多臺(tái)服務(wù)器的集群模式。無(wú)論哪種部署方式，修改Kafka的配置文件perties都是至關(guān)重要的。單機(jī)模式和集群模式部署的步驟基本一致，只是在perties文件的配置上有些差異。7.3.3Kafka配置文件$KAFKA_HOME/config中有多個(gè)配置文件。perties配置參數(shù)（部分）參數(shù)名說明broker.id用于指定Broker服務(wù)器對(duì)應(yīng)的ID，各個(gè)服務(wù)器的值不同listeners表示監(jiān)聽的地址及端口，PLAINTEXT表示純文本，也就是說，不管發(fā)送什么數(shù)據(jù)類型都以純文本的方式接收，包括圖片，視頻等work.threads網(wǎng)絡(luò)線程數(shù)，默認(rèn)是3num.io.threadsI/O線程數(shù)，默認(rèn)是8socket.send.buffer.bytes套接字發(fā)送緩沖，默認(rèn)是100KBsocket.receive.buffer.bytes套接字接收緩沖，默認(rèn)是100KBsocket.request.max.bytes接收到的最大字節(jié)數(shù)，默認(rèn)是100MBlog.dirs用于指定Kafka數(shù)據(jù)存放目錄，地址可以是多個(gè)，多個(gè)地址需用逗號(hào)分割num.partitions分區(qū)數(shù)，默認(rèn)是1num.recovery.threads.per.data.dir每一個(gè)文件夾的恢復(fù)線程，默認(rèn)是1log.retention.hours數(shù)據(jù)保存時(shí)間，默認(rèn)是168h，即一個(gè)星期（7天）log.segment.bytes指定每個(gè)數(shù)據(jù)日志保存最大數(shù)據(jù)，默認(rèn)為1GB，當(dāng)超過這個(gè)值時(shí)，會(huì)自動(dòng)進(jìn)行日志滾動(dòng)erval.ms設(shè)置日志過期的時(shí)間，默認(rèn)為300s（即5min）zookeeper.connect用于指定Kafka所依賴的ZooKeeper集群的IP和端口號(hào)，地址可以是多個(gè)，多個(gè)地址需用逗號(hào)分割zookeeper.connection.timeout.ms設(shè)置Zookeeper的連接超時(shí)時(shí)間，默認(rèn)為6s，如果到達(dá)這個(gè)指定時(shí)間仍然連接不上就默認(rèn)該節(jié)點(diǎn)發(fā)生故障7.4KafkaShell常用命令Kafka支持的所有命令在$KAFKA_HOME/bin下存放。7.4KafkaShell常用命令命令功能描述kafka-server-start.sh啟動(dòng)KafkaBrokerkafka-server-stop.sh關(guān)閉KafkaBrokerkafka-topics.sh創(chuàng)建、刪除、查看、修改Topickafka-console-producer.sh啟動(dòng)Producer，生產(chǎn)消息，從標(biāo)準(zhǔn)輸入讀取數(shù)據(jù)并發(fā)布到Kafkakafka-console-consumer.sh啟動(dòng)Consumer，消費(fèi)消息，從Kafka讀取數(shù)據(jù)并輸出到標(biāo)準(zhǔn)輸出7.5綜合實(shí)戰(zhàn)：Kafka實(shí)戰(zhàn)規(guī)劃Kafka集群部署Kafka集群?jiǎn)?dòng)Kafka集群驗(yàn)證Kafka集群使用KafkaShell關(guān)閉Kafka集群7.5.1規(guī)劃Kafka集群主機(jī)名IP地址運(yùn)行服務(wù)軟硬件配置master30QuorumPeerMainKafka內(nèi)存：4GCPU：1個(gè)2核硬盤：40G操作系統(tǒng)：CentOS7.6.1810Java：OracleJDK8u191ZooKeeper：ZooKeeper3.4.13Kafka：Kafka2.1.1slave131QuorumPeerMainKafka內(nèi)存：1GCPU：1個(gè)1核硬盤：20G操作系統(tǒng)：CentOS7.6.1810Java：OracleJDK8u191ZooKeeper：ZooKeeper3.4.13Kafka：Kafka2.1.1slave232QuorumPeerMainKafka內(nèi)存：1GCPU：1個(gè)1核硬盤：20G操作系統(tǒng)：CentOS7.6.1810Java：OracleJDK8u191ZooKeeper：ZooKeeper3.4.13Kafka：Kafka2.1.17.5.1規(guī)劃Kafka集群軟件名稱軟件版本發(fā)布日期下載地址VMwareWorkstationProVMwareWorkstation14.5.7ProforWindows2017年6月22日/products/workstation-pro.htmlCentOSCentOS7.6.18102018年11月26日/download/JavaOracleJDK8u1912018年10月16日/technetwork/java/javase/downloads/index.htmlZooKeeperZooKeeper3.4.132018年7月15日/releases.htmlKafkaKafka2.1.12019年2月15日/downloads7.5.2部署Kafka集群1.初始軟硬件環(huán)境準(zhǔn)備（1）準(zhǔn)備三臺(tái)機(jī)器，安裝操作系統(tǒng)，編者使用CentOSLinux7。（2）對(duì)集群內(nèi)每一臺(tái)機(jī)器，配置靜態(tài)IP、修改機(jī)器名、添加集群級(jí)別域名映射、關(guān)閉防火墻。（3）對(duì)集群內(nèi)每一臺(tái)機(jī)器，安裝和配置Java，要求Java8或更高版本，編者使用OracleJDK8u191。（4）安裝和配置Linux集群中各節(jié)點(diǎn)間的SSH免密登錄。（5）在Linux集群上部署ZooKeeper集群。7.5.2部署Kafka集群2.獲取KafkaKafka官方下載地址為/downloads，編者選用的Kafka版本是2019年2月15日發(fā)布的Kafka2.1.1，其安裝包文件kafka_2.12-2.1.1.tgz例如存放在master機(jī)器的/home/xuluhui/Downloads中。讀者應(yīng)該注意到了，Kafka安裝包和一般安裝包的命名方式不一樣，例如kafka_2.12-2.1.1.tgz，其中2.12是Scala版本，2.1.1才是Kafka版本，官方強(qiáng)烈建議Scala版本和服務(wù)器上的Scala版本保持一致，避免引發(fā)一些不可預(yù)知的問題。3.安裝Kafka以下所有操作需要在三臺(tái)機(jī)器上完成。切換到root，解壓kafka_2.12-2.1.1.tgz到安裝目錄如/usr/local下，使用命令如下所示。[xuluhui@master~]$suroot[root@masterxuluhui]#cd/usr/local[root@masterlocal]#tar-zxvf/home/xuluhui/Downloads/kafka_2.12-2.1.1.tgz7.5.2部署Kafka集群4.配置Kafka修改Kafka配置文件perties，master機(jī)器上的配置文件$KAFKA_HOME/config/perties修改后的幾個(gè)參數(shù)如下所示。broker.id=0log.dirs=/usr/local/kafka_2.12-2.1.1/kafka-logszookeeper.connect=master:2181,slave1:2181,slave2:2181slave1和slave2機(jī)器上的配置文件$KAFKA_HOME/config/perties中參數(shù)broker.id依次設(shè)置為1、2，其余參數(shù)值與master機(jī)器相同。5.創(chuàng)建所需目錄以上第4步驟使用了系統(tǒng)不存在的目錄：Kafka數(shù)據(jù)存放目錄/usr/local/kafka_2.12-2.1.1/kafka-logs，因此需要?jiǎng)?chuàng)建它，使用的命令如下所示。[root@masterlocal]#mkdir/usr/local/kafka_2.12-2.1.1/kafka-logs7.5.2部署Kafka集群6.設(shè)置$KAFKA_HOME目錄屬主為了在普通用戶下使用Kafka，將$KAFKA_HOME目錄屬主設(shè)置為L(zhǎng)inux普通用戶例如xuluhui，使用以下命令完成。[root@masterlocal]#chown-Rxuluhui/usr/local/kafka_2.12-2.1.17.在系統(tǒng)配置文件目錄/etc/profile.d下新建kafka.sh使用“vim/etc/profile.d/kafka.sh”命令在/etc/profile.d文件夾下新建文件kafka.sh，添加如下內(nèi)容。exportKAFKA_HOME=/usr/local/kafka_2.12-2.1.1exportPATH=$KAFKA_HOME/bin:$PATH其次，重啟機(jī)器，使之生效。此步驟可省略。7.5.3啟動(dòng)Kafka集群首先，在三臺(tái)機(jī)器上使用命令“zkServer.shstart”啟動(dòng)ZooKeeper集群，確保其正常運(yùn)行。其次，在三臺(tái)機(jī)器上使用以下命令啟動(dòng)Kafka，此處以master機(jī)器為例。[xuluhui@master~]$kafka-server-start.sh-daemon$KAFKA_HOME/config/perties這里需要注意的是，啟動(dòng)腳本若不加-daemon參數(shù)，則如果執(zhí)行Ctrl+Z后會(huì)退出，且啟動(dòng)的進(jìn)程也會(huì)退出，所以建議加-daemon參數(shù)，實(shí)現(xiàn)以守護(hù)進(jìn)程方式啟動(dòng)。7.5.4驗(yàn)證Kafka集群方法：jps7.5.5使用KafkaShell【案例7-1】使用Kafka命令創(chuàng)建Topic、查看Topic，啟動(dòng)Producer生產(chǎn)消息，啟動(dòng)Consumer消費(fèi)消息。【案例7-1】（1）創(chuàng)建Topic在任意一臺(tái)機(jī)器上創(chuàng)建Topic“kafkacluster-test”。[xuluhui@master~]$kafka-topics.sh--create\--zookeepermaster:2181,slave1:2181,slave2:2181\--replication-factor3\--partitions3\--topickafkacluster-test由于共部署了三個(gè)Broker，所以創(chuàng)建Topic時(shí)能指定--replication-factor3。其中，選項(xiàng)--zookeeper用于指定ZooKeeper集群列表，可以指定所有節(jié)點(diǎn)，也可以指定為部分節(jié)點(diǎn)；選項(xiàng)--replication-factor為復(fù)制數(shù)目，數(shù)據(jù)會(huì)自動(dòng)同步到其他Broker上，防止某個(gè)Broker宕機(jī)數(shù)據(jù)丟失；選項(xiàng)--partitions用于指定一個(gè)Topic可以切分成幾個(gè)partition，一個(gè)消費(fèi)者可以消費(fèi)多個(gè)partition，但一個(gè)partition只能被一個(gè)消費(fèi)者消費(fèi)?！景咐?-1】（2）查看Topic詳情在任意一臺(tái)機(jī)器上查看Topic“kafkacluster-test”的詳情。Topic“kafkacluster-test”總計(jì)有3個(gè)分區(qū)（PartitionCount），副本數(shù)為3（ReplicationFactor），且每個(gè)分區(qū)上有3個(gè)副本（通過Replicas的值可以得出），另外最后一列Isr（In-SyncReplicas）表示處理同步狀態(tài)的副本集合，這些副本與Leader副本保持同步，沒有任何同步延遲。另外，Leader、Replicas、Isr中的數(shù)字就是BrokerID，對(duì)應(yīng)配置文件config/perties中的broker.id參數(shù)值?！景咐?-1】（3）啟動(dòng)生產(chǎn)者生產(chǎn)消息在master機(jī)器上使用kafka-console-producer.sh啟動(dòng)生產(chǎn)者，使用命令如下所示。[xuluhui@master~]$kafka-console-producer.sh\--broker-listmaster:9092,slave1:9092,slave2:9092\--topickafkacluster-test（4）啟動(dòng)消費(fèi)者消費(fèi)消息在slave1和slave2機(jī)器上分別使用kafka-console-consumer.sh啟動(dòng)消費(fèi)者，以slave1機(jī)器為例，使用命令如下所示。[xuluhui@slave1~]$kafka-console-consumer.sh\--bootstrap-servermaster:9092,slave1:9092,slave2:9092\--topickafk