import bdqn.sparkdemp.util.ChenIpUtil
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
object UDFDemo {
def main(args: Array[String]): Unit = {
// 自定义一个函数，实现查询dep表时，deptno返回的是具体的部门信息不是编号信息
// 使用UDF来实现
val session: SparkSession = SparkSession.builder()
.master("local")
.appName(this.getClass.getSimpleName)
.getOrCreate()
val ipDS: Dataset[String] = session.read.textFile("D:\\testLogs\\ip.txt")
val logDS: Dataset[String] = session.read.textFile("D:\\testLogs\\ipaccess.log")
import session.implicits._
val arrRules: Array[(Long, Long, String)] = ipDS.map(t => {
val str: Array[String] = t.split("\\|")
(str(2).toLong, str(3).toLong, str(6))
}).collect()
val longIpDF: DataFrame = logDS.map(str => {
// 把ip转换成一个long
ChenIpUtil.ip2long(str.split("\\|")(1))
}).toDF("longIp")
longIpDF.createTempView("t_longIp")
// 自定义一个UDF函数，实现，接受一个long值，返回一个省份
// 使用UDF，需要先注册
session.udf.register("ip2Pro", (ip: Long) => {
Binarysearch(arrRules, ip)
})
// 使用自定义函数
session.sql("select longIp,ip2Pro(longIp) from t_longIp").show()
}
// 二分查找
def Binarysearch(arr: Array[(Long, Long, String)], ip: Long): String = {
var left = 0
var right = arr.size - 1
while (left <= right) {
var mid = left + ((right - left) >> 1) // 防止溢出， 并且移位操作更高效
if (arr(mid)._1 > ip) {
right = mid - 1
} else if (arr(mid)._2 < ip) {
left = mid + 1
} else {
return arr(mid)._3 //注意 添加return， scala中只有最后一行才是作为返回值
}
}
null
}
}
UDFDemo
©著作权归作者所有,转载或内容合作请联系作者
- 文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
- 文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
- 文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
