Kafka命令行的利用/Spark-Streaming核心编程（二）

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核心编程（二）

• 需求：通过 SparkStreaming 从 Kafka 读取数据，并将读取过来的数据做简单计算，最终打印到控制台。
  
• 导入依赖
  

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

• 编写代码
  

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()
  }
}

• 开启Kafka集群
  

• 开启Kafka生产者，产生数据
  

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

• 运行程序，接收Kafka生产的数据并进行相应处置惩罚
  

8）查看消费进度
kafka-consumer-groups.sh --describe --bootstrap-server node01:9092,node02:9092,node03:9092 --group kafka

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。