sparksql系列(三) sparksql列操作、窗口函数、join

一：Sparksql列操作

初始化SparkContext及数据：

import java.util.Arrays

import org.apache.spark.SparkConf
import org.apache.spark.api.java.JavaSparkContext
import org.apache.spark.sql.{DataFrame, Row, SparkSession, functions}
import org.apache.spark.sql.functions.{col, desc, length, row_number, trim, when}
import org.apache.spark.sql.functions.{countDistinct,sum,count,avg}
import org.apache.spark.sql.functions.concat
import org.apache.spark.sql.types.{LongType, StringType, StructField, StructType}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.storage.StorageLevel
import org.apache.spark.sql.SaveMode
import java.util.ArrayList

object WordCount {

　　def initSparkAndData() : DataFrame = {

　　　　val sparkSession= SparkSession.builder().master("local").appName("AppName").getOrCreate()
　　　　val javasc = new JavaSparkContext(sparkSession.sparkContext)
　　　　val nameRDD = javasc.parallelize(Arrays.asList("{‘name‘:‘wangwu‘,‘age‘:‘18‘,‘vip‘:‘t‘}",
　　　　　　"{‘name‘:‘sunliu‘,‘age‘:‘19‘,‘vip‘:‘t‘}","{‘name‘:‘zhangsan‘,‘age‘:‘20‘,‘vip‘:‘f‘}"));
　　　　val namedf = sparkSession.read.json(nameRDD)

　　　　namedf
　　}
}

增加列

　　　　val data = initSparkAndData()

　　　　//方法一：可以添加常量值
　　　　data.select(when(col("name").isNotNull,1).otherwise(0) as "usergroup").show(100)
　　　　//方法二：只能已经存在的列操作
　　　　data.withColumn("time", concat(col("age"),col("name")) ).show(100)

删除列

　　　　val data = initSparkAndData()
　　　　data.drop("vip").show(100)

二：SparkSql 窗口函数

　　　　传统数据库中就有这个函数，就是partation by () order by ()。那下面让我们看看sparksql中怎么写：

　　　　val data = initSparkAndData()
　　　　data.withColumn("isVsip", row_number().over(Window.partitionBy(col("vip")).orderBy(desc("name")))).show(100)

　　　　上面的意思是按照VIP分组，后按照name排序，作为新的列isVsip。项目中用来作为提取最新记录的函数，举例如下：

　　　　统计出用户最近登录记录：

　　　　val sparkSession= SparkSession.builder().master("local").appName("AppName").getOrCreate()
　　　　val javasc = new JavaSparkContext(sparkSession.sparkContext)

　　　　val nameRDD1 = javasc.parallelize(Arrays.asList("{‘name‘:‘wangwu‘,‘time‘:‘2019-08-12‘}",
　　　　　　"{‘name‘:‘sunliu‘,‘time‘:‘2019-08-13‘}","{‘name‘:‘zhangsan‘,‘time‘:‘2019-08-14‘}"));
　　　　val namedf1 = sparkSession.read.json(nameRDD1)

　　　　val nameRDD2 = javasc.parallelize(Arrays.asList("{‘name‘:‘wangwu‘,‘time‘:‘2019-09-12‘}",
　　　　　　"{‘name‘:‘sunliu‘,‘time‘:‘2019-08-13‘}","{‘name‘:‘zhangsan‘,‘time‘:‘2019-07-14‘}"));
　　　　val namedf2 = sparkSession.read.json(nameRDD2)
　　　　//上面全是构造数据。
　　　　namedf1.union(namedf2).withColumn("max_time", row_number().over(Window.partitionBy(col("name")).orderBy(desc("time"))))
　　　　　　.filter(col("max_time") ===1).show(100)

三：Sparksql join操作

初始化SparkContext及数据：

import java.util.Arrays

object WordCount {
　　def joinTestData() = {
　　　　val sparkSession= SparkSession.builder().master("local").appName("AppName").getOrCreate()
　　　　val javasc = new JavaSparkContext(sparkSession.sparkContext)

　　　　val nameRDD = javasc.parallelize(Arrays.asList("{‘name‘:‘zhangsan‘,‘age‘:‘18‘,‘sex‘:‘N‘}", "{‘name‘:‘lisi‘,‘age‘:‘19‘,‘sex‘:‘F‘}","{‘‘:‘‘,‘‘:‘‘,‘‘:‘‘}"));
　　　　val nameRDD1 = javasc.parallelize(Arrays.asList("{‘name‘:‘wangwu‘,‘age‘:‘18‘,‘vip‘:‘t‘}", "{‘name‘:‘sunliu‘,‘age‘:‘19‘,‘vip‘:‘t‘}","{‘name‘:‘zhangsan‘,‘age‘:‘18‘,‘vip‘:‘f‘}"));
　　　　val data1 = sparkSession.read.json(nameRDD)
　　　　val data2 = sparkSession.read.json(nameRDD1)

　　　　(data1,data2)
　　}
}