大橙子网站建设,新征程启航
为企业提供网站建设、域名注册、服务器等服务
这篇文章主要讲解了“怎么自定义JDBCRDD的分区”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“怎么自定义JDBCRDD的分区”吧!
岳普湖网站制作公司哪家好,找创新互联!从网页设计、网站建设、微信开发、APP开发、响应式网站等网站项目制作,到程序开发,运营维护。创新互联从2013年成立到现在10年的时间,我们拥有了丰富的建站经验和运维经验,来保证我们的工作的顺利进行。专注于网站建设就选创新互联。
1,JDBCRDD使用
val data = new JdbcRDD(sc, getConnection
, "SELECT id,aa FROM bbb where ? <= ID AND ID <= ?", lowerBound = 3, upperBound =5, numPartitions = 1, mapRow = extractValues)
参数解释:
1,sparkcontext。
2,一个创建链接的函数。
3,sql。必须有? <= ID AND ID <= ?。
4,要取数据的id最小行。
5,要取数据的id最大行号。
6,分区数。
7,一个将ResultSet转化为需要类型的方法。
2,JdbcRDD的getPartition方法
override def getPartitions: Array[Partition] = {
// bounds are inclusive, hence the + 1 here and - 1 on end
val length = BigInt(1) + upperBound - lowerBound
(0 until numPartitions).map(i => {
val start = lowerBound + ((i * length) / numPartitions)
val end = lowerBound + (((i + 1) * length) / numPartitions) - 1
new JdbcPartition(i, start.toLong, end.toLong)
}).toArray
}
3,JdbcRDD的compute方法
就是一个通过jdbc获取指定范围数据的过程。
val part = thePart.asInstanceOf[JdbcPartition]
val conn = getConnection()
val stmt = conn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)
stmt.setLong(1, part.lower)
stmt.setLong(2, part.upper)
val rs = stmt.executeQuery()
4,重写JDBC方法
重写分区的方法即可。
如:
CustomizedJdbcRDD[T: ClassTag](
sc: SparkContext,
getConnection: () => Connection,
sql: String,
getCustomizedPartitions: () => Array[Partition],
prepareStatement: (PreparedStatement, CustomizedJdbcPartition) => PreparedStatement,
mapRow: (ResultSet) => T = CustomizedJdbcRDD.resultSetToObjectArray _)
同时把getPartition方法重写为:
override def getPartitions: Array[Partition] = {
getCustomizedPartitions();
}
感谢各位的阅读,以上就是“怎么自定义JDBCRDD的分区”的内容了,经过本文的学习后,相信大家对怎么自定义JDBCRDD的分区这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是创新互联,小编将为大家推送更多相关知识点的文章,欢迎关注!