怎么自定義JDBCRDD的分區

這篇文章主要講解了“怎么自定義JDBCRDD的分區”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“怎么自定義JDBCRDD的分區”吧！

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的分區這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！