Kafka命令行的使用

創建topic

kafka-topics.sh --create --zookeeper node01:2181,node02:2181,node03:2181 --topic test1 --partitions 3 --replication-factor 3

分區數量，副本數量，都是必須的。

數據的形式：

主題名稱-分區編號。

在Kafka的數據目錄下查看。

設定副本數量，不能大于broker的數量。

2.2查看所有的topic

kafka-topics.sh --list --zookeeper node01:2181,node02:2181,node03:2181

2.3查看某個topic的詳細信息

kafka-topics.sh --describe --zookeeper node01:2181,node02:2181,node03:2181?--topic test1

ISR:?In-Sync Replicas ??可以提供服務的副本。

AR = ISR + OSR

2.4刪除topic

kafka-topics.sh --delete --zookeeper node01:2181,node02:2181,node03:2181?--topic test1

2.5生產數據

kafka-console-producer.sh：

指定broker

指定topic

寫數據的命令：

kafka-console-producer.sh --broker-list node01:9092,node02:9092,node03:9092 --topic test1

Spark-Streaming核心編程（二）

1. 需求：通過 SparkStreaming 從 Kafka 讀取數據，并將讀取過來的數據做簡單計算，最終打印到控制臺。
2. 導入依賴

<dependency>
????<groupId>org.apache.spark</groupId>
????<artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
????<version>3.0.0</version>
</dependency>

1. 編寫代碼

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010._
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.streaming.dstream.{DStream, InputDStream}

object DirectAPI {
??def main(args: Array[String]): Unit = {
????val sparkConf = new SparkConf().setMaster("local[*]").setAppName("direct")

????val ssc = new StreamingContext(sparkConf, Seconds(3))

????// 定義?Kafka 相關參數
????val kafkaPara: Map[String, Object] = Map[String, Object](
??????ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node01:9092,node02:9092,node03:9092",
??????ConsumerConfig.GROUP_ID_CONFIG -> "kafka",
??????"key.deserializer" -> classOf[StringDeserializer],
??????"value.deserializer" -> classOf[StringDeserializer]
????)

????// 通過讀取?Kafka 數據，創建?DStream
????val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
??????ssc,
??????LocationStrategies.PreferConsistent,
??????ConsumerStrategies.Subscribe[String, String](Set("kafka"), kafkaPara)
????)

????// 提取出數據中的?value 部分
????val valueDStream: DStream[String] = kafkaDStream.map(record => record.value())

????// wordCount 計算邏輯
????valueDStream.flatMap(_.split(" "))
??????.map((_, 1))
??????.reduceByKey(_ + _)
??????.print()

????ssc.start()
????ssc.awaitTermination()
??}
}

1. 開啟Kafka集群

1. 開啟Kafka生產者，產生數據

kafka-console-producer.sh --broker-list node01:9092,node02:9092,node03:9092?--topic kafka

1. 運行程序，接收Kafka生產的數據并進行相應處理

8）查看消費進度

kafka-consumer-groups.sh --describe --bootstrap-server node01:9092,node02:9092,node03:9092 --group kafka