亲宝软件园·资讯

展开

Spark JDBC操作MySQL方式详细讲解

CarveStone 人气:0

JDBC操作MySQL

在实际的企业级开发环境中,如果数据规模特S别大,此时采用传统的SQL语句去处理的话一般需要分成很多批次处理,而且很容易造成数据库服务宕机,且实际的处理过程可能会非常复杂,通过传统的Java EE等技术可能很难或者不方便实现处理算法,此时采用SparkSQL进行分布式分析处理就可以非常好的解决该问题,在生产环境下,一般会在Spark SQL和具体要操作的DB之间加上一个缓冲层次,例如中间使用Redis或者Kafka。

Spark SQL可以通过JDBC从传统的关系型数据库中读写数据,读取数据后直接生成的是DataFrame,然后再加上借助于Spark SQL丰富的API来进行各种操作。从计算数据规模的角度去讲,集群并行访问数据库数据,调用Data Frame Reader的Format(“JDBC”)的方式说明Spark SQL操作的数据来源是通过JDBC获得,JDBC后端一般都是数据库,例如MySQL、Oracle等。

JDBC读取数据方式

单Partition(无并发)

调用函数格式:def jdbc(url: String, table: String, properties: Properties): DataFrame

这种方法是将所有的数据放在一个Partition中进行操作(即并发度为1),意味着无论给的资源有多少,只有一个Task会执行任务,执行效率比较慢,并且容易出现OOM。使用如下,在spark-shell中执行:

/*此为代码格式,实际中使用应替换相应字段中的内容*/
val url = "jdbc:mysql://localhost:/database"        
val tableName = "table"
// 设置连接用户&密码
val prop = new java.util.Properties
prop.setProperty("user","username")  //实际使用中替换username为相应的用户名
prop.setProperty("password","pwd")   //实际使用中替换pwd为相应的密码

根据Long类型字段分区

/*此为代码格式,实际中使用应替换相应字段中的内容*/
def jdbc(
url: String,
table: String,
columnName: String, // 根据该字段分区,需要为整型,比如 id 等
lowerBound: Long, // 分区的下界
upperBound: Long, // 分区的上界
numPartitions: Int,  //分区的个数
connectionProperties: Properties): DataFrame

根据字段将数据进行分区,放进不同的Partition中,执行效率较快,但是只能根据数据字段作为分区关键字。使用如下:

/*此为代码格式,实际中使用应替换相应字段中的内容*/
val url = "jdbc:mysql://mysqlHost:3306/database"
val tableName = "table"
val columnName = "colName"
val lowerBound = 1,
val upperBound = 10000000,
val numPartitions = 10,
// 设置连接用户&密码
val prop = new java.util.Properties
prop.setProperty("user","username")
prop.setProperty("password","pwd")

将字段 colName 中发 1~10000000 条数据分区到 10 个 Partition 中。

根据任意类型字段分区

/*此为代码格式,实际中使用应替换相应字段中的内容*/
jdbc(
url: String,
table: String,
predicates: Array[String],
connectionProperties: Properties): DataFrame

以下使用时间字段进行分区:

/*此为代码格式,实际中使用应替换相应字段中的内容*/
val url = "jdbc:mysql://mysqlHost:3306/database"
val tableName = "table"
// 设置连接用户&密码
val prop = new java.util.Properties
prop.setProperty("user","username")
prop.setProperty("password","pwd")
/**
* 将 9 月 16-12 月 15 三个月的数据取出,按时间分为 6 个 partition
* 为了减少事例代码,这里的时间都是写死的
* modified_time 为时间字段
*/
val predicates =
Array(
"2015-09-16" -> "2015-09-30",
"2015-10-01" -> "2015-10-15",
"2015-10-16" -> "2015-10-31",
"2015-11-01" -> "2015-11-14",
"2015-11-15" -> "2015-11-30",
"2015-12-01" -> "2015-12-15"
).map {
case (start, end) =>
s"cast(modified_time as date) >= date '$start' " + s"AND cast(modified_time
as date) <= date '$end'"
}

这种方法可以使用任意字段进行分区,比较灵活,适用于各种场景。以MySQL 3000W数据量为例,如果单分区count,若干分钟就会报OOM;如果分成5~20个分区后,count操作只需要2s,效率会明显提高,这里就凸显出JDBC高并发的优势。Spark高并发度可以大幅度提高读取以及处理数据的速度,但是如果设置过高(大量的Partition同时读取)也可能会将数据源数据库宕掉。

JDBC读取MySQL数据

下面来进行实际操作,首先需要配置MySQL

修改表格的权限,目的是为了使其他主机可以远程连接 MySQL,通过此命令可以查看访问用户允许的主机名。

通过命令修改host为%,表示任意IP地址都可以登录。出现ERROR 1062 (23000): Duplicate entry '%-root' for key 'PRIMARY',是因为 user+host 是主键,不能重复,可以不用理会。也可通过以下命令删除user 为空的内容来解决:delete from user where user='';

在MySQL创建数据库和表格,插入数据,查看:

create database test;    //创建数据库test
use test;        //进入数据库test
create table people( name varchar(12), age int);         //创建表格people并构建结构
insert into people values ("Andy",30),("Justin",19),("Dela",25),("Magi",20),("Pule",21),("Mike",12);        //向people表中插入数据
select * from people;         //输出people表中全部数据

编写代码读取MySQL表中数据:

//导入依赖环境
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext}
import java.util.Properties
val url = "jdbc:mysql://localhost/test"       //MySQL地址及数据库
val username = "root"       //用户名
val sqlContext = new SQLContext(sc)       
sc.setLogLevel("WARN")
val uri = url + "?user=" + username + "&useUnicode=true&characterEncoding=UTF-8"           //设置读取路径及用户名
val properties = new Properties()      //创建JDBC连接信息
properties.put("user","root")
properties.put("driver", "com.mysql.jdbc.Driver")
val df_test: DataFrame = spark.sqlContext.read.jdbc(uri, "people", properties)  //读取数据
df_test.select("name","age").collect().foreach(row => {         //输出数据
	println("name " + row(0) + ", age" + row(1))
})
df_test.write.mode("append").jdbc(uri,"people",properties)     //向people表中写入读出的数据,相当于people表中有两份一样的数据

加载全部内容

相关教程
猜你喜欢
用户评论