这篇文章将为大家详细讲解有关怎么理解spark的自定义分区和排序及spark与jdbc，文章内容质量较高，因此小编分享给大家做个参考，希望大家阅读完这篇文章后对相关知识有一定的了解。
//自定义分区
import org.apache.spark.SparkConf
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.Partitioner
object PrimitivePartitionTest {
def main(args: Array[String]): Unit = {
val conf = new SparkConf
conf.setMaster("local[2]").setAppName("Partitioner")
val context = new SparkContext(conf)
val rdd = context.parallelize(List(("hgs",2),("wd",44),("cm",99),("zz",100),("xzhh",67)), 2)
//实例化类，并设置分区类
val partitioner = new CustomPartitioner(2)
val rdd1 = rdd.partitionBy(partitioner)
rdd1.saveAsTextFile("c://partitioner")
context.stop()
}
}
//自定义分区类继承spark的Partitioner
class CustomPartitioner(val partitions:Int ) extends Partitioner{
def numPartitions: Int= this.partitions
def getPartition(key: Any): Int={
if(key.toString().length()<=2)
0
else
1
}
}
//自定义排序
package hgs.spark.othertest
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import scala.math.Ordered
//自定义排序第一种实现方式，通过继承ordered
class Student(val name:String,var age:Int) extends Ordered[Student] with Serializable{
def compare(that: Student): Int={
return this.age-that.age
}
}
class Boy(val name:String,var age:Int) extends Serializable{
}
//第二种方式通过实现隐式转换实现
object MyPredef{
implicit def toOrderBoy = new Ordering[Boy]{
def compare(x: Boy, y: Boy): Int={
x.age - y.age
}
}
}
//引入隐式转换
import MyPredef._
object CutstomOrder {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
conf.setMaster("local[2]").setAppName("CutstomOrder")
val context = new SparkContext(conf)
val rdd = context.parallelize(List(("hgs",2),("wd",44),("cm",99),("zz",100),("xzhh",67)), 2)
//下面的第二个参数false为降序排列
//val rdd_sorted = rdd.sortBy(f=>new Student(f._1,f._2), false, 1)
val rdd_sorted = rdd.sortBy(f=>new Boy(f._1,f._2), false, 1)
rdd_sorted.saveAsTextFile("d://ordered")
context.stop()
}
}
//JDBC
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.rdd.JdbcRDD
import java.sql.Connection
import java.sql.DriverManager
import java.sql.ResultSet
import scala.collection.mutable.ListBuffer
object DataFromJdbcToSpark {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
conf.setMaster("local[2]").setAppName("BroadCastTest")
val context = new SparkContext(conf)
val sql = "select name,age from test where id>=? and id <=?"
var list = new ListBuffer[(String,Int)]()
//第七个参数是一个自定义的函数，spark会调用该函数，完成自定义的逻辑，y的数据类型是ResultSet,该函数不可以想自己定义的数组添加数据，
//应为应用的函数会将结果保存在JdbcRDD中
val jdbcRDD = new JdbcRDD(context,getConnection,sql,1,8,2,y=>{
(y.getString(1),y.getInt(2))
})
println(jdbcRDD.collect().toBuffer)
context.stop()
}
def getConnection():Connection={
Class.forName("com.mysql.jdbc.Driver")
val conn = DriverManager.getConnection("jdbc:mysql://192.168.6.133:3306/hgs","root","123456");
conn
}
}
//----------------------------------------------------------------------
package hgs.spark.othertest
import java.sql.Connection
import java.sql.DriverManager
import org.apache.commons.dbutils.QueryRunner
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
//将spark计算后的结果录入数据库
object DataFromSparktoJdbc {
def main(args: Array[String]): Unit = {
val conf = new SparkConf
conf.setMaster("local[2]").setAppName("DataFromSparktoJdbc")
val context = new SparkContext(conf)
val addressrdd= context.textFile("d://words")
val words = addressrdd.flatMap(_.split(" ")).map(x=>(x,1)).reduceByKey(_+_)
//println(words.partitions.size)
var p:Int =0
words.foreachPartition(iter=>{
//每个分区一个链接
val qr = new QueryRunner()
val conn = getConnection
println(conn)
val sql = s"insert into words values(?,?)"
//可以修改为批量插入效率更高
while(iter.hasNext){
val tpm = iter.next()
val obj1 :Object = tpm._1
val obj2 :Object = new Integer(tpm._2)
//obj1+conn.toString()可以看到数据库的插入数据作用有三个不同的链接
qr.update(conn, sql,obj1+conn.toString(),obj2)
}
//println(conn)
//println(p)
conn.close()
})
words.saveAsTextFile("d://wordresult")
}
def getConnection():Connection={
Class.forName("com.mysql.jdbc.Driver")
val conn = DriverManager.getConnection("jdbc:mysql://192.168.6.133:3306/hgs","root","123456");
conn
}
}
//广播变量
package hgs.spark.othertest
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
object BroadCastTest{
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
conf.setMaster("local[2]").setAppName("BroadCastTest")
val context = new SparkContext(conf)
val addressrdd= context.textFile("d://address")
val splitaddrdd = addressrdd.map(x=>{
val cs = x.split(",")
(cs(0),cs(1))
}).collect().toMap
//广播变量，数据被缓存在每个节点，减少了节点之间的数据传送，可以有效的增加效率，广播出去的可以是任意的数据类型
val maprdd = context.broadcast(splitaddrdd)
val namerdd = context.textFile("d://name")
val result = namerdd.map(x=>{
//该出使用了广播的出去的数组
maprdd.value.getOrElse(x, "UnKnown")
})
println(result.collect().toBuffer)
context.stop()
}
}
其他一些知识点
1.spark 广播变量 rdd.brodcastz(rdd)，广播变量的用处是将数据汇聚传输到各个excutor上面
，这样在做数据处理的时候减少了数据的传输
2.wordcount程序
context.textFile(args(0),1).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
wordcount程序代码，一个wordcount会产生5个RDD
sc.textFile() 会产生两个RDD 1.HadoopRDD-> MapPartitionsRDD
flatMap() 会产生MapPartitionsRDD
map 会产生MapPartitionsRDD
reduceByKey 产生ShuuledRDD
saveAsTextFile
3.缓存数据到内存 rdd.cache 清理缓存 rdd.unpersist(true),rdd.persist存储及级别 cache方法调用的是persist方法
4.spark 远程debug，需要设置sparkcontext.setMaster("spark://xx.xx.xx.xx:7077").setJar("d:/jars/xx.jar")
关于怎么理解spark的自定义分区和排序及spark与jdbc就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。
原创文章，作者：ItWorker，如若转载，请注明出处：https://blog.ytso.com/222774.html