看注释

  1. package cool.xiaokang.sql
  3. import org.apache.spark.sql.{DataFrame, SparkSession}
  4. import org.apache.spark.sql.functions._
  6. /**
  7.  * SparkSql常用函数
  8.  *
  9.  * @author: xiaokang
  10.  *        微信公众号：小康新鲜事儿
  11.  *        小康个人文档：https://www.xiaokang.cool/
  12.  */
  13. // 1000    xiaokangXXs聆听ゝ尔伈    916458603   蔡文姬 123.234.77.79   微信  2020-03-20 09:30:11 2020-03-20
  14. case class UserLogin(loginId: Int, userName: String, userId: String, roleName: String, loginIp: String, loginType: String, loginTime: String, loginDate: String)
  15. object SparkSqlDemo {
  16.   // 创建sparkSession
  17.   def createSparkSession(master: String = "local[*]", appName: String = this.getClass.getName, isHive: Boolean = false): SparkSession = {
  18.     if (isHive) {
  19.       SparkSession.builder().master(master).appName(master).enableHiveSupport().getOrCreate()
  20.     } else {
  21.       SparkSession.builder().master(master).appName(appName).getOrCreate()
  22.     }
  23.   }
  24.   // 创建DataFrame
  25.   def readDataSource(path: String, isRdd: Boolean = true): (DataFrame,SparkSession) = {
  26.     val session = createSparkSession()
  27.     session.sparkContext.setLogLevel("WARN")
  28.     if (isRdd) {
  29.       val dataSourceRDD = session.sparkContext.textFile(path)
  30.       val userLoginRDD = dataSourceRDD.mapPartitions(it => {
  31.         it.map { line => {
  32.           val userStr = line.split("\t")
  33.           UserLogin(userStr(0).toInt, userStr(1), userStr(2), userStr(3), userStr(4), userStr(5), userStr(6), userStr(7))
  34.         }
  35.         }
  36.       })
  37.       import session.implicits._
  38.       (userLoginRDD.toDF(),session)
  39.     } else {
  40.       (session.read.text(path),session)
  41.     }
  42.   }
  43.   def main(args: Array[String]): Unit = {
  44.     val userLoginDf = readDataSource("D:/IDEA-Projects/SparkApp/game_login_data")
  45.     userLoginDf._1.where("userId=916458603")
  46.       //当前日期基础上加1天
  47.     .withColumn("date_add",date_add(col("loginTime"),1))
  48.       //当前日期基础上减1天
  49.       .withColumn("date_sub",date_sub(col("loginTime"),1))
  50.       //获取当前日期的下一个星期一的日期
  51.       .withColumn("next_day",next_day(col("loginTime"),"MON"))
  52.       //获取当月最后一天日期
  53.       .withColumn("last_day",last_day(col("loginTime")))
  54.       //获取年
  55.       .withColumn("year",year(col("loginTime")))
  56.       //获取月
  57.       .withColumn("month",month(col("loginTime")))
  58.       //获取季度
  59.       .withColumn("quarter",quarter(col("loginTime")))
  60.       //一年中第几天
  61.       .withColumn("dayOfYear",dayofyear(col("loginTime")))
  62.       //获取时间戳
  63.       .withColumn("unixTimestamp",unix_timestamp(col("loginTime")))
  64.       //日期截取
  65.       .withColumn("parseToDay",substring(col("loginTime"),1,10))
  66.       //字符串拼接
  67.       .withColumn("concat",concat(col("userName"),col("userId")))
  68.       //字符串拼接并指定拼接时候的拼接符
  69.       .withColumn("concat_ws",concat_ws("***",col("userName"),col("userId")))
  70.       //将每个单词的首字母变为大写
  71.       .withColumn("initcap",initcap(col("userName")))
  72.       //转小写
  73.       .withColumn("lower",lower(col("userName")))
  74.       //转大写
  75.       .withColumn("upper",upper(col("userName")))
  76.       .show()
  77.   }
  78. }