13518219792
本篇内容主要讲解“Spark-Streaming如何处理数据到MySQL中”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“Spark-Streaming如何处理数据到mysql中”吧!
我们注重客户提出的每个要求，我们充分考虑每一个细节，我们积极的做好网站设计、网站建设服务，我们努力开拓更好的视野，通过不懈的努力，创新互联赢得了业内的良好声誉，这一切，也不断的激励着我们更好的服务客户。 主要业务：网站建设,网站制作,网站设计,小程序开发,网站开发,技术开发实力，DIV+CSS，PHP及ASP，ASP.Net，SQL数据库的技术开发工程师。
数据表如下：
create database test; use test; DROP TABLE IF EXISTS car_gps; CREATE TABLE IF NOT EXISTS car_gps( deployNum VARCHAR(30) COMMENT '调度编号', plateNum VARCHAR(10) COMMENT '车牌号', timeStr VARCHAR(20) COMMENT '时间戳', lng VARCHAR(20) COMMENT '经度', lat VARCHAR(20) COMMENT '纬度', dbtime TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '数据入库时间', PRIMARY KEY(deployNum, plateNum, timeStr))
首先引入mysql的驱动
mysql mysql-connector-java 5.1.44
package com.hoult.Streaming.work
import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util.Properties
import com.hoult.structed.bean.BusInfo
import org.apache.spark.sql.ForeachWriter
class JdbcHelper extends ForeachWriter[BusInfo] {
var conn: Connection = _
var statement: PreparedStatement = _
override def open(partitionId: Long, epochId: Long): Boolean = {
if (conn == null) {
conn = JdbcHelper.openConnection
}
true
}
override def process(value: BusInfo): Unit = {
//把数据写入mysql表中
val arr: Array[String] = value.lglat.split("_")
val sql = "insert into car_gps(deployNum,plateNum,timeStr,lng,lat) values(?,?,?,?,?)"
statement = conn.prepareStatement(sql)
statement.setString(1, value.deployNum)
statement.setString(2, value.plateNum)
statement.setString(3, value.timeStr)
statement.setString(4, arr(0))
statement.setString(5, arr(1))
statement.executeUpdate()
}
override def close(errorOrNull: Throwable): Unit = {
if (null != conn) conn.close()
if (null != statement) statement.close()
}
}
object JdbcHelper {
var conn: Connection = _
val url = "jdbc:mysql://hadoop1:3306/test?useUnicode=true&characterEncoding=utf8"
val username = "root"
val password = "123456"
def openConnection: Connection = {
if (null == conn || conn.isClosed) {
val p = new Properties
Class.forName("com.mysql.jdbc.Driver")
conn = DriverManager.getConnection(url, username, password)
}
conn
}
}package com.hoult.Streaming.work
import com.hoult.structed.bean.BusInfo
import org.apache.spark.sql.{Column, DataFrame, Dataset, SparkSession}
object KafkaToJdbc {
def main(args: Array[String]): Unit = {
System.setProperty("HADOOP_USER_NAME", "root")
//1 获取sparksession
val spark: SparkSession = SparkSession.builder()
.master("local[*]")
.appName(KafkaToJdbc.getClass.getName)
.getOrCreate()
spark.sparkContext.setLogLevel("WARN")
import spark.implicits._
//2 定义读取kafka数据源
val kafkaDf: DataFrame = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "linux121:9092")
.option("subscribe", "test_bus_info")
.load()
//3 处理数据
val kafkaValDf: DataFrame = kafkaDf.selectExpr("CAST(value AS STRING)")
//转为ds
val kafkaDs: Dataset[String] = kafkaValDf.as[String]
//解析出经纬度数据，写入redis
//封装为一个case class方便后续获取指定字段的数据
val busInfoDs: Dataset[BusInfo] = kafkaDs.map(BusInfo(_)).filter(_ != null)
//将数据写入MySQL表
busInfoDs.writeStream
.foreach(new JdbcHelper)
.outputMode("append")
.start()
.awaitTermination()
}
}kafka-topics.sh --zookeeper linux121:2181/myKafka --create --topic test_bus_info --partitions 2 --replication-factor 1 kafka-console-producer.sh --broker-list linux121:9092 --topic test_bus_info
到此，相信大家对“Spark-Streaming如何处理数据到mysql中”有了更深的了解，不妨来实际操作一番吧！这里是创新互联网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！
