vlambda博客
学习文章列表

uid,dtguid01,2018-02-28guid01,2018-03-01guid01,2018-03-01guid01,2018-03-02guid01,2018-03-05guid01,2018-03-04guid01,2018-03-06guid01,2018-03-07guid02,2018-03-01guid02,2018-03-02guid02,2018-03-03guid02,2018-03-06

 //创建SparkSession val spark = SparkSession.builder() .appName(this.getClass.getSimpleName) .master("local[*]") .getOrCreate()  //纯sql进行查询数据 val df1: DataFrame = spark .read .option("header", "true") .csv("data1.txt")  df1.createTempView("tb_log")  //uid01,2018-03-01 // 第一种实现方式写sql val df2 = spark.sql( """ | |select |uid, |min(dt) as min_dt, |max(dt) as max_dt, |count(date_diff) as times |from |(select |uid, |dt, |date_sub(dt,dt_num) as date_diff | from | ( | select | uid, | dt, | row_number() over(partition by uid order by dt asc) as dt_num | from | ( | select | distinct(uid,dt),uid,dt | from tb_log | )t1 | )t2) | group by uid,date_diff having times>=3 |""".stripMargin).show()

 //第二种方式 使用DSL风格的代码实现 import spark.implicits._ import org.apache.spark.sql.functions._ df1.distinct(). select('uid, 'dt, (row_number() over (Window.partitionBy("uid").orderBy("dt"))) as 'rn ) .select( 'uid, 'dt, date_sub('dt, 'rn) as 'date_diff ).groupBy('uid, 'date_diff) //假如要多个聚合时 使用agg .agg( min("dt"), max("dt"), count("*") as "times" ).where('times >= 2) .drop("date_diff") .show()

/** * @author:tom * @Date:Created in 9:42 2021/1/5 */object AccumulateDemo { Logger.getLogger("org").setLevel(Level.WARN)  def main(args: Array[String]): Unit = {  //创建SparkSession val spark = SparkSession.builder() .appName(this.getClass.getSimpleName) .master("local[*]") .getOrCreate()  //纯sql进行查询数据 val df1: DataFrame = spark .read .option("header", "true") .csv("shop.csv")  df1.createTempView("v_shop")  spark.sql( s""" |select |sid, |mth, |sum(mth_money) over(partition by sid order by mth) as total_money |from |( |select |sid, |mth, |sum(money) as mth_money |from |( |select |sid, |date_format(dt,"yyyy-MM") as mth, |cast(money as double) as money |from v_shop |) t1 group by sid,mth) t2 | |""".stripMargin).show()

 //dsl风格编程 import spark.implicits._ import org.apache.spark.sql.functions._ df1.select($"sid", 'money.cast(DataTypes.DoubleType) as "money", expr("date_format(dt, 'yyyy-MM') as mth") ).groupBy("sid", "mth"). sum("money") .withColumnRenamed("sum(money)", "mth_money") .select( $"sid", $"mth", sum("mth_money").over(Window.partitionBy("sid") .orderBy("mth")) as "total_money" ).show()

/** * @author:tom * @Date:Created in 19:41 2021/1/5 */object FlowDemo { Logger.getLogger("org").setLevel(Level.WARN)  def main(args: Array[String]): Unit = { //创建SparkSession val spark = SparkSession.builder() .appName(this.getClass.getSimpleName) .master("local[*]") .getOrCreate()  //uid,start_dt,end_dt,flow //1,2020-02-18 14:20:30,2020-02-18 14:46:30,20  //纯sql进行查询数据 val df1: DataFrame = spark .read .option("header", "true") .csv("flow.txt")  df1.createTempView("v_flow")  spark.sql( """ | |select | uid, | min(start_dt) as start_dt, | max(end_dt) as end_dt, | sum(flow) as flow |from |( |select |uid, |start_dt, |end_dt, |sum(lag_num) over(partition by uid order by start_dt)as flag, |flow |from |( |select |uid, |start_dt, |end_dt, |if((to_unix_timestamp(start_dt)-to_unix_timestamp(lag_time))/60>10,1,0) as lag_num, |flow |from |( |select |uid, |start_dt, |end_dt, |flow, |lag(end_dt,1,start_dt) over(partition by uid order by start_dt) as lag_time |from v_flow |)t1 )t2 )t3 group by uid,flag |""".stripMargin).show()

 import spark.implicits._ import org.apache.spark.sql.functions._ //dsl风格代码 df1.select( $"uid", $"start_dt", $"end_dt", $"flow", expr("lag(end_dt) over(partition by uid order by start_dt) as lag_time") ).select( $"uid", $"start_dt", $"end_dt", $"flow", expr("if((to_unix_timestamp(start_dt)-to_unix_timestamp(lag_time))/60>10,1,0) as lag_num") ).select( $"uid", $"start_dt", $"end_dt", $"flow", sum("lag_num").over(Window.partitionBy("uid") .orderBy("start_dt")) as "flag" ).groupBy("uid","flag"). agg( min("start_dt") as "start_dt", max("end_dt") as "end_dt", sum("flow") as "flow" ).drop("flag") .orderBy("uid") .show()