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

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

ConsumerGroup2

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

Broker1Topic1Topic2Topic3Topic4

Broker2Topic1Topic2Topic3

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