Kafka簡介消息系統負責將數據從一個應用程序傳輸到另一個應用程序，因此應用程序可以專注于數據，但不擔心如何共享它。

分布式消息傳遞基于可靠消息隊列的概念。

消息在客戶端應用程序和消息傳遞系統之間異步排隊。

有兩種類型的消息模式可用-一種是點對點，另一種是發布-訂閱(pub-sub)消息系統。

什么是消息系統什么是消息系統解決了什么痛點?為什么引入Kafka？ApacheKafka是一個分布式發布-訂閱消息系統和一個強大的隊列，可以處理大量的數據，并使您能夠將消息從一個端點傳遞到另一個端點。

Kafka適合離線和在線消息消費。

Kafka消息保留在磁盤上，并在群集內復制以防止數據丟失。

Kafka構建在ZooKeeper同步服務之上。

它與ApacheStorm和Spark可以非常好地集成，用于實時流式數據分析。Kafka簡介為什么引入Kafka？可靠性-Kafka是分布式，分區，復制和容錯的。可擴展性-Kafka消息傳遞系統輕松縮放，無需停機。分布式-日志的分區partition（分布）在Kafka集群的服務器上。耐用性-Kafka使用分布式提交日志，這意味著消息會盡可能快地保留在磁盤上，因此它是持久的。性能-Kafka對于發布和訂閱消息都具有高吞吐量。即使存儲了許多TB的消息，它也保持穩定的性能。Kafka特性Kafka架構Kafka核心概念Kafka數據流對于每一個topic，Kafka集群都會維持一個分區日志。每個分區都是有序且順序不可變的記錄集，并且不斷地追加到結構化的commitlog文件。分區中的每一個記錄都會分配一個id來表示順序，也叫offset。offset用來唯一的標識分區中每一條記錄。Kafka數據流Kafka集群保留所有發布的記錄—無論他們是否已被消費—并通過一個可配置的參數——保留期限來控制。例如，如果保留策略設置為100天（當然不建議大家這樣做），一條記錄發布后100天內，可以隨時被消費，100天后這條記錄會被拋棄并釋放磁盤空間。Kafka安裝配置-JDK&ZookeeperUbuntu16，1CPU，1GMemory，10GDisk，VirtualSystemHostname:IPAddress:00User/Pass:jacob/jacobLinux連接工具環境準備Linux環境準備下載JDK安裝包并上傳到Ubuntu下載Zookeeper并上傳到Ubuntu安裝包準備JDK&Zookeeper安裝包1.解壓縮：tar–zxf2.修改配置文件：sudovim/etc/profile3.使配置文件生效:source/etc/profile5.檢查安裝包準備安裝JDK1.解壓縮：tar–zxf2.修改配置文件：sudovim/etc/profile3.使配置文件生效:source/etc/profile5.修改Zookeeper配置文件安裝包準備安裝Zookeeper1.ls2.create/set3.get4.deleteZookeeper常用操作Kafka安裝配置-單代理配置及操作Ubuntu16，1CPU，1GMemory，10GDisk，VirtualSystemJDKZookeeper環境準備Linux環境準備1.下載Kafka安裝包2.解壓安裝包3.修改配置文件環境準備Kafka安裝包Kafka常用操作Kafka啟動與關閉1.Producer創建Topicbin/kafka-console-producer.sh--broker-listnode100:9092--topicHello_WorldKafka常用操作KafkaProducer1.Consumer讀取Topicbin/kafka-console-consumer.sh--bootstrap-servernode100:9092--topicHello_World--from-beginningKafka常用操作KafkaConsumerKafka安裝配置–多代理配置及操作1臺Ubuntu配置多個Brokers3臺Ubuntu配置多個Brokers，每臺Ubuntu一個Broker配置說明1臺UbuntuVS3臺Ubuntu三臺Ubuntu16，1CPU，1GMemory，10GDisk，VirtualSystemHosts：node100,node101,node102全部配置JDK配置Zookeeper集群環境準備Linux環境準備1.克隆node100到node101,node1022.修改機器名和IP配置環境準備Linux環境準備1.修改zoo.cfg配置文件2.修改myid配置文件3.啟動環境準備Zookeeper配置tickTime=2000initLimit=10syncLimit=5dataDir=/home/jacob/apache-zookeeper-3.5.5-bin/zookeeper_dataclientPort=2181server.1=node100:2888:3888server.2=node101:2888:3888server.3=node102:2888:38881.修改perties配置文件2.清空data目錄3.啟動環境準備Kafka配置1.創建Topic(2partitions&2replication-factor)bin/kafka-topic.sh--bootstrap-servernode100:9092--partitionsinteger--replication-factorinteger--topictopic-nameKafka常用操作KafkaTopcis1.創建Topic(2partitions&3replication-factor)bin/kafka-topic.sh--bootstrap-servernode100:9092--partitionsinteger--replication-factorinteger--topictopic-nameKafka常用操作KafkaTopcis1.創建Topic(3partitions&3replication-factor)bin/kafka-topic.sh--bootstrap-servernode100:9092--partitionsinteger--replication-factorinteger--topictopic-nameKafka常用操作KafkaTopcis1.Producer生產Topic./kafka-console-producer.sh--broker-listnode100:9092,node101:9092,node102:9092--topictest_02_02Kafka常用操作KafkaProducer&&ConsumerTopictest_02_02有多少分區？這些數據如何分布？如何讀取這些數據1.Consumer讀取Topic./kafka-console-consumer.sh--bootstrap-servernode100:9092--topictest_02_02--partition0--from-beginning./kafka-console-consumer.sh--bootstrap-servernode100:9092--topictest_02_02--partition1--from-beginningKafka常用操作KafkaProducer&&ConsumerProducer加多一行數據會被哪個partition讀取？如何讀取所有數據？1.Consumer讀取Topic./kafka-console-consumer.sh--bootstrap-servernode100:9092--topictest_02_02--from-beginningKafka常用操作KafkaProducer&&Consumer如果是topictest_03_03呢？1.Topictest_02_02./kafka-topics.sh--bootstrap-servernode100:9092--describe--topictest_02_02Kafka常用操作Kafka容錯性驗證如果broker1丟失會如何？數據還能訪問么？Kafka開發環境配置&ProducerAPI開發環境說明IntellijIDEA&&Maven&&JDK&&Git<properties><piler.target>1.8</piler.target><piler.source>1.8</piler.source></properties>Pom配置CompilerConfiguration<build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.6.1</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin></plugins></build><!--/artifact/org.apache.kafka/kafka-clients--><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.3.1</version></dependency>Pom配置KafkaDependencyKafkaProducerAPIProducer&KafkaProducerclass源碼分析ClientID?MessageSend?KafkaProducerAPIProducerRecordclass源碼分析ProducerRecord控制哪些變量？KafkaProducer實例化的四種方式ProducerRecord實例化的六種方式KafkaProducerAPIKafkaProducer實戰實戰一發送消息到test_02_02|V消息|KV消息實戰二發送消息到test_02_02|指定PartitionKafkaConsumerAPIKafkaConsumerAPIConsumer&KafkaConsumerclass源碼分析GroupID?ReceiveMessage?KafkaProducerAPIConsumerRecords&ConsumerRecordclass源碼分析ConsumerRecord控制哪些變量？KafkaConsumer實例化ConsumerRecords&ConsumerRecord實例化KafkaConsumerAPIKafkaConsumer實戰實戰一接收從test_02_02的消息|V消息|KV消息實戰二指定Partition接收test_02_02消息設計一個工具類可以返回隨機字符串設計Producer可以每秒發送數據設計Consumer可以讀取數據KafkaProducer&ConsumerProducer和Consumer合并實戰實戰三發送和消費test_02_02消息Kafka整合SparkHadoop&Spark環境說明Hadoop環境Hadoop&Spark環境說明Hadoop常用操作Hadoop&Spark環境說明Spark環境Hadoop&Spark環境說明Spark常用操作Kafka整合Spark原理&架構SparkStreamingKafka整合Spark原理&架構Kafka整合SparkAPP1APP2BrokerBrokerBrokerKafkaClusterSparkStreamingSparkEngineHDFSTerminalKafka整合Spark開發環境配置AddScalaFrameworkSupport<dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.11</artifactId><version>2.4.3</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.11</artifactId><version>2.4.3</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-8_2.11</artifactId><version>2.4.3</version></dependency>Kafka整合Spark開發環境配置AddMavenDependencies&開發流程

本地IDEA開發本地代碼本地測試編譯打包上傳Jar包集群運行Kafka整合Spark實戰Kafka整合Spark實戰SparkStreamingAPI如何實例化StreamingContextval

conf=newSparkConf().setMaster("local[2]").setAppName("NetworkWordCount")val

ssc=newStreamingContext(conf,Seconds(1))ssc.start()//Startthecomputation

ssc.awaitTermination()//WaitforthecomputationtoterminateTransformationKafka整合Spark實戰SparkStreamingKafkaAPIvallines=KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafka