<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.sparkstream</groupId><artifactId>LyhSparkStreaming</artifactId><version>1.0-SNAPSHOT</version><properties><spark.version>2.3.3</spark.version><scala.version>2.11</scala.version></properties><dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_${scala.version}</artifactId><version>${spark.version}</version><exclusions><exclusion><groupId>commons-beanutils</groupId><artifactId>commons-beanutils-core</artifactId></exclusion></exclusions></dependency><!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka_2.11</artifactId><version>1.6.3</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_${scala.version}</artifactId><version>${spark.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_${scala.version}</artifactId><version>${spark.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-hive_${scala.version}</artifactId><version>${spark.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-mllib_${scala.version}</artifactId><version>${spark.version}</version></dependency></dependencies><build><plugins><plugin><groupId>org.scala-tools</groupId><artifactId>maven-scala-plugin</artifactId><version>2.15.2</version><executions><execution><goals><goal>compile</goal><goal>testCompile</goal></goals></execution></executions></plugin><plugin><artifactId>maven-compiler-plugin</artifactId><version>3.6.0</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-surefire-plugin</artifactId><version>2.19</version><configuration><skip>true</skip></configuration></plugin></plugins></build></project>

package KafkaAndStreamingimport java.io.{BufferedReader, FileInputStream, FileNotFoundException, IOException, InputStreamReader}
import java.util.Propertiesimport org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.clients.producer.{Callback, KafkaProducer, Producer, ProducerRecord, RecordMetadata}object TestKafkaProducer {type Key = Stringtype Val = Stringdef getProducerCnfig():Properties={/*** 对于kafka producer的相关配置文件项* 还有其他的属性，自己去查一下**/val props:Properties = new Properties()// Kafka 的 urlprops.put("bootstrap.servers", "localhost:9092")// group.id 随便写props.put("group.id", "producer-group")props.put("replication.factor", "min.insync.replicas")// 备份数量props.put("min.insync.replicas", "3")props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")props}def main(args: Array[String])={// 获取配置文件val props:Properties = getProducerCnfig()// 创建生产者var producer = new KafkaProducer[String, String](props)try{//读取保存的文件val fis:FileInputStream = new FileInputStream("/Users/hunter/text1.txt")val isr:InputStreamReader = new InputStreamReader(fis, "UTF-8")val br:BufferedReader = new BufferedReader(isr)var line: String = ""line = br.readLine()var i: Int=0while (line != null) {producer.send(toMessage(line.toString, Option(i.toString), Option("Hunter")),new Callback {override def onCompletion(recordMetadata: RecordMetadata, e: Exception): Unit = {println(s"""Message $i, send to: """ + recordMetadata.topic())}})i += 1line = br.readLine()}producer.close()br.close()isr.close()fis.close()} catch {case ex: FileNotFoundException =>{println("Missing file exception")}case ex: IOException => {println("IO Exception")}case _ =>{println("Have other Error")}}
//    dealWithData}//把我的消息包装成了ProducerRecordprivate def toMessage(value: String, key: Option[Key] = None, topic: Option[String] = None): ProducerRecord[Key, Val] = {val t = topic.getOrElse(Option("test").getOrElse(throw new IllegalArgumentException("Must provide topic or default topic")))require(!t.isEmpty, "Topic must not be empty")key match {case Some(k) => new ProducerRecord(t, k, value)case _ => new ProducerRecord(t, value)}}
}

package KafkaAndStreamingimport kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.spark.{SparkConf, TaskContext}
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}object KafkaAndPrintInSpark {//判断设置的时输入参数，是否包含brokers 和 topic 至少参数的长度为2，即单机运行一个test的topic：   broker=localhost:9092 topic=testdef main(args: Array[String]) {if (args.length < 2) {System.err.println(s"""|Usage: DirectKafkaWordCount <brokers> <topics>|  <brokers> is a list of one or more Kafka brokers|  <topics> is a list of one or more kafka topics to consume from""".stripMargin)System.exit(1)}//将参数args读入到数组中val Array(brokers, topics) = args// 用2秒批间隔创建上下文val sparkConf = new SparkConf().setMaster("local[2]").setAppName("DirectKafkaWordCount")val ssc = new StreamingContext(sparkConf, Seconds(5))// 创建kafka流与brokers和topicval topicsSet = topics.split(",").toSetval kafkaParams = Map[String, String]("metadata.broker.list" -> brokers,"bootstrap.servers" -> "localhost:9092",
//      "auto.offset.reset" -> "smallest","key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer","value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer")val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)// 此处注释部分是自定义偏移量 5L代表从 5开始读取。默认是读取最新的，offset从上一次读取结束的位置开始
//    val offsetList = List(("Hunter", 0, 5L))
//    val fromOffsets = setFromOffsets(offsetList)//对List进行处理，变成需要的格式，即Map[TopicAndPartition, Long]
//    val messageHandler = (mam: MessageAndMetadata[String, String]) => (mam.topic, mam.message()) //构建MessageAndMetadata，这个所有使用情况都是一样的，就这么写
//    val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](
//      ssc, kafkaParams, fromOffsets, messageHandler)messages.foreachRDD( rdd => {
//      if(rdd.count()>0) {rdd.foreach( records => {println("_1: " + records._1)println("_2: " + records._2)})val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRangesrdd.foreachPartition { iter =>val o: OffsetRange = offsetRanges(TaskContext.get.partitionId())println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")}
//      }})// 开始计算ssc.start()ssc.awaitTermination()}def setFromOffsets(list: List[(String, Int, Long)]): Map[TopicAndPartition, Long] = {var fromOffsets: Map[TopicAndPartition, Long] = Map()for (offset <- list) {val tp = TopicAndPartition(offset._1, offset._2)//topic和分区数fromOffsets += (tp -> offset._3)           // offset位置}fromOffsets}
}