工作中需要对dataframe 写出非常复杂的处理函数
使用udf 对单个函数进行处理,使之可以对整列数据进行处理。
示例一: 对两列数据求cos(x,y)
这个是实际应用的代码
两个问题
1.dataframe 里面的数据是vector格式的,儿哟啊进行处理需要处理成Array格式的
2.对dataframe 进行两列处理,需要用UDF处理,udf函数的输入是两个同类型列
def cosineDistence(ve1: Vetor, Ve2:Vector): Double = {
val v1 = ve1.toArray
val v2 = ve2.toArray
var distance = -1.0
val vector =v1.zip(v2)
val x1x2 = vetor.map{ case (x1, x2) => x1 *x2}.sum
var x1sum = v1.map(x1 =>math.pow(x1, 2)).sum
x1sum = math.pow(x1sum, 1.0/2)
var x2sum = v2.map(x2 => math.pow(x2,2)).sum
x2sum = math.pow(x2sum, 1.0/2)
distance = x1x2 / (x1sum * x2sum)
distance}
val udf_consineDistance = udf(consineDistence _)
val outcomes = output.limit(1).select($"feature" as "one").crossJoin(output)
val outcomes = outcomes.withColumn("test_result", udf_consineDistance($"one",$"feature"))
示例二:特定词频统计
四个问题:
1.如何对udf函数设置多个参数,使用Curring 函数法则进行处理,同时一定小心书写格式,需要在等号左边添加上自定义参数
2.函数输出多个值,需要在函数定义处定义好输出,输出值应该是对应的
3.使用Option[T]进行函数设计,防止程序运行崩溃,应该正确的使用map
4.对dataframe进行处理的时候需要注意udf的使用规则,这里面只有一个输入列
5.在dataframe 使用filter判断是否相等,应用“===”
6.一定注意split(".")和split('.’)
def get_set(num: Int) :(Set[String], Set[String])={
val input = scala.io.Source.fromFile(s"/home/zhuyin/words/${num}_line").getlines().toList
val set1 = input(0).split(",").map(_.trim).toSet
val set2 = input(1).split(",").map(_.trim).toSet
(set1,set2)
}
(set1,set2) = get_set(1)
def KeywordsCount(set1: Set[String],Set2: Set[String])(str :String):Option[Int]={
Option(str).map{ s=>
val pro = s.split(" ").toSet
val samewords1 = pro& set1
val samewords2 = pro& set2
val final1 = if(result.isEmpty) 0 else 1
final1
}
}
def CountAll(set1: Set[String], set2: Set[String])(str : String) : Option[Int] ={
Option(str).map{ s =>
val pro = s.split('.')
val count = 0
for(s<- pro){
if(KeywordsCount(set1,set2)(s) == 1){count = 1}
}
count
}
}
def udf_KeywordCount(set1: Set[string], set2:Set[String]) = udf(CountAll(set1, sert2) _)
val df_final = df.withColumn("count1",udf_KeywordCount(set1,set2)($"descrip"))
df_final.filter(df_final("count1")===1).select("count1").count().toInt