spark中生成时间序列数据的函数stack和sequence

Sequence函数

用Sequence函数生成时间序列函数，真的是非常简便易用，之前因为没找到，所以走了不少弯路。

    println("指定开始和结束数字，生成对应的数字序列，通过第三个参数来控制步长")
    SparkUtil.executeSQL("""
                         |select explode(sequence(1,10,2)) id
                         |""".stripMargin)(spark)
1
2
3
4

    println("指定开始和结束日期，生成对应的日期序列，默认递增1天")
    SparkUtil.executeSQL("""
                           |select explode(sequence(to_date('2022-11-01','yyyy-MM-dd'),to_date('2022-11-10','yyyy-MM-dd'))) dt
                           |""".stripMargin)(spark)
1
2
3
4

    println("指定开始和结束日期，生成对应的日期序列，指定递增间隔天数")
    SparkUtil.executeSQL("""
                           |select explode(sequence(to_date('2022-11-01','yyyy-MM-dd'),to_date('2022-11-10','yyyy-MM-dd'),interval 2 day)) dt
                           |""".stripMargin)(spark)
1
2
3
4

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-2JDnvpDa-1669168562419)(https://img2022.cnblogs.com/blog/367203/202211/367203-20221120234501000-1380321919.png)]

    println("指定开始和结束小时日期，生成小时级别的时间序列(小时级别需要转成时间戳)")
    SparkUtil.executeSQL("""
                           |select explode(sequence(to_timestamp('2022-11-01 01:00:00','yyyy-MM-dd HH:mm:ss'),to_timestamp('2022-11-04 01:00:00','yyyy-MM-dd HH:mm:ss'),interval 1 hours)) dt
                           |""".stripMargin,100)(spark)
1
2
3
4

    println("指定开始和结束小时时间戳，生成小时级别的时间序列，增量直接指定毫秒")
    println("开始毫秒时间戳 1667235600000 -> 2022-11-01 01:00:00")
    println("结束毫秒时间戳 1667581200000 -> 2022-11-05 01:00:00")
    SparkUtil.executeSQL("""
                           |select explode(sequence(1667235600000,1667581200000,3600000)) dt
                           |""".stripMargin,100)(spark)
1
2
3
4
5
6

    SparkUtil.executeSQL("""
                    |
                    |with tmp_a as (
                    | select '2022-11-01' as id,'01,02,03' as infos
                    | union all
                    | select '2022-11-02' as id,'04,05,06' as infos
                    |)
                    |select id,info from tmp_a LATERAL VIEW explode(split(infos,',')) t AS info
                    |""".stripMargin)(spark)
1
2
3
4
5
6
7
8
9

在spark中向前填充或向后填充的案例，这个其实就是先排下，然后用last或者first取值去替换。
其中last(timeField, true)第二个参数，是是否忽略空值。

    val window = Window.partitionBy(idField).orderBy(timeField).rowsBetween(-1, 0)
    val filled = last(timeField, true).over(window)
    outputDF = outputDF.selectExpr(field1: _*).withColumn(rule.getField, filled)
    //后向填充
    val window = Window.partitionBy(idField).orderBy(timeField).rowsBetween(0, 1)
    val filled = last(timeField, true).over(window)
    outputDF = outputDF.withColumn(rule.getField, filled)
1
2
3
4
5
6
7

stack 将x列中以y为组拆分为N行

一共x1-x5一共5个字段，每2个y1,y2名称作为一行，，拆分为N=3行。
如果x确定的，y也确定的，那么n也是确定的，不能随便写。

stack(N,x1,x2,x3,x4,x5) as (y1,y2)
1

例如下面的：就是将5列(x1-x5）以(field1,field2)拆分为3行。

stack(3,'x1','x2','x3','x4','x5') as (field1,field2)
1

+-------+-------+
|symbol1|symbol2|
+-------+-------+
|     x1|     x2|
|     x3|     x4|
|     x5|   null|
+-------+-------+
1
2
3
4
5
6
7

但是如果一共5个字段，每两个字段作为一行，拆分为2行，这个就很明显不行了，会报错。
如果只有4个字段x1-x4，那么刚好可以拆分为2行，每行2个字段。

stack(2,'x1','x2','x3','x4','x5') as (field1,field2)
1

: 'The number of aliases supplied in the AS clause does not match the number of columns output by the UDTF expected 3 aliases but got symbol1,symbol2

    SparkUtil.executeSQL(
      """
        |select stack(1, 1, 2, 3) as (dt1,dt2,dt3)
        |""".stripMargin)(spark)
1
2
3
4

    SparkUtil.executeSQL(
      """
        |select stack(2, 1, 2, 3) as (dt1,dt2)
        |""".stripMargin)(spark)
1
2
3
4

    SparkUtil.executeSQL(
      """
        |select stack(3, 1, 2, 3) as dt1
        |""".stripMargin)(spark)
1
2
3
4

    SparkUtil.executeSQL("""
        |select stack(3, split('1,2,3,4,5',',')) as dt1
        |""".stripMargin)(spark)
1
2
3