2023_Spark_实验三十二：消费Kafka数据并保存到MySQL中

实验目的：掌握Scala开发工具消费Kafka数据，并将结果保存到关系型数据库中

实验方法：消费Kafka数据保存到MySQL中

实验步骤：

一、创建Job_ClickData_Process

代码如下：

package examsimport org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies, OffsetRange}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}import java.sql.{Connection, DriverManager, PreparedStatement}
import scala.collection.mutable/*** @projectName sparkGNU2023  * @package exams  * @className exams.Job_ClickData_Process  * @description ${description}  * @author pblh123* @date 2023/12/20 15:42* @version 1.0**/object Job_ClickData_Process {def main(args: Array[String]): Unit = {//  1. 创建spark,sc,sparkstreaming对象if (args.length != 3) {println("您需要输入三个参数")System.exit(5)}val musrl: String = args(0)val conf = new SparkConf().setAppName(s"${this.getClass.getSimpleName}").setMaster(musrl)val sc: SparkContext = new SparkContext(conf)sc.setLogLevel("WARN")val ckeckpointdir: String = args(1)val ssc = new StreamingContext(sc, Seconds(5)) //连续流批次处理的大小//  2. 代码主体
//    设置ckeckpoint目录ssc.checkpoint(ckeckpointdir)//准备kafka的连接参数val kfkbst: String = args(2)val kafkaParams: Map[String, Object] = Map[String, Object]("bootstrap.servers" -> kfkbst,"group.id" -> "SparkKafka",//latest表示如果记录了偏移量的位置，就从记录的位置开始消费，如果没有记录，就从最新/或最后的位置开始消费//earliest表示如果记录了偏移量的位置，就从记录的位置开始消费，如果没有记录，就从最开始/最早的位置开始消费//none示如果记录了偏移量的位置，就从记录的位置开始消费，如果没有记录，则报错"auto.offset.reset" -> "latest", //偏移量的重置位置"enable.auto.commit" -> (false: java.lang.Boolean), //是否自动提交偏移量"key.deserializer" -> classOf[StringDeserializer],"value.deserializer" -> classOf[StringDeserializer])val topics: Array[String] = Array("RealDataTopic")//从mysql中查询出offsets:Map[TopicPartition, Long]val offsetsMap: mutable.Map[TopicPartition, Long] = OffsetUtils.getOffsetMap("SparkKafka", "RealDataTopic")val kafkaDS: InputDStream[ConsumerRecord[String, String]] = if (offsetsMap.size > 0) {println("MySql记录了offset信息,从offset处开始消费")//连接kafka的消息KafkaUtils.createDirectStream[String, String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String, String](topics, kafkaParams, offsetsMap))} else {println("MySql没有记录了offset信息,从latest处开始消费")//连接kafka的消息KafkaUtils.createDirectStream[String, String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String, String](topics, kafkaParams))}//实时处理数据并手动维护offsetval valueDS = kafkaDS.map(_.value()) //_表示从kafka中消费出来的每一条数据valueDS.print()kafkaDS.map(_.value())valueDS.foreachRDD(rdd => {rdd.foreachPartition(lines => {//将处理分析的结果存入mysql/*DROP TABLE IF EXISTS `job_real_time`;CREATE TABLE `job_real_time` (`datetime` varchar(8) DEFAULT NULL COMMENT '日期',`job_type` int(2) DEFAULT NULL COMMENT '1代表新招聘岗位，0代表找工作的人',`job_id` int(8) DEFAULT NULL COMMENT '岗位ID，匹配岗位名称',`count` int(8) DEFAULT NULL COMMENT '企业新增岗位数和找工作的人数') ENGINE=InnoDB DEFAULT CHARSET=utf8;*///1.开启连接val conn: Connection = DriverManager.getConnection("jdbc:mysql://192.168.137.110:3306/bigdata19?characterEncoding=UTF-8&serverTimezone=UTC", "lh", "Lh123456!")//2.编写sql并获取psval sql: String = "replace into job_real_time(datetime,job_type,job_id,count) values(?,?,?,?)"val ps: PreparedStatement = conn.prepareStatement(sql)//3.设置参数并执行for (line <- lines) {var item = line.split(" ")ps.setString(1, item(0).toString)ps.setInt(2, item(1).toInt)ps.setInt(3, item(2).toInt)ps.setInt(4, item(3).toInt)ps.executeUpdate()}//4.关闭资源ps.close()conn.close()})})//手动提交偏移量kafkaDS.foreachRDD(rdd => {if (rdd.count() > 0) {//获取偏移量val offsets: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRangesOffsetUtils.saveOffsets(groupId = "SparkKafka", offsets)}})//开启sparkstreaming任务并等待结束，关闭ssc,scssc.start()ssc.awaitTermination()ssc.stop()sc.stop()}}