更新于：

模块B 1.4

第4小题

本题比较难，代码如下：

抽取 ds_db01 库中 coupon_use 的增量数据进入Hive的 ods 库中表 coupon_use ，增量字段取 ods.coupon_use 表中 get_time、used_time、pay_time 中的最大者，只将新增的数据抽入，字段名称、类型不变，同时添加静态分区，分区字段为 etl_date ，类型为 String，且值为当前比赛日的前一天日期（分区字段格式为 yyyyMMdd）。使用Hive Cli 查询最新分区数据总条数，将执行结果截图粘贴至客户端桌面【Release\模块 B 提交结果.docx】中对应的任务序号下；

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
import java.util.Properties

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession

object test {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("akka").setLevel(Level.OFF)
    Logger.getLogger("org").setLevel(Level.OFF)

    val spark=SparkSession.builder().appName("aaa")
      .enableHiveSupport()
      .config("hive.metastore.uris","thrift://192.168.152.82:9083")
      .getOrCreate()

    val properties=new Properties()
    properties.setProperty("user","root")
    properties.setProperty("password","123456")

    val df01=spark.read.jdbc("jdbc:mysql://192.168.152.82:3306/ds_db01","coupon_use",properties)
    df01.show(5,false)
    df01.createOrReplaceTempView("t_df01")

    //对于NULL空值处理特别重要，稍有不慎，满盘皆输，要观察好再写入
    def check01(a:AnyVal,b:AnyVal,c:AnyVal):String={
      var result="XXX"
      var a1=a.toString
      var b1=b.toString
      var c1=c.toString
      if (a1=="NULL") a1="0"
      if (b1=="NULL") b1="0"
      if (c1=="NULL") c1="0"
      if (a1.toInt>b1.toInt) result=a1
      else result=b1
      if (result.toInt<c1.toInt) result=c1
      result
    }
    spark.udf.register("check01",check01 _)

    val df02=spark.sql("select *,check01(get_time,used_time,pay_time) as max_time from ods.coupon_use")
    df02.show(5)
    df02.createOrReplaceTempView("t_df02")

    val df03=spark.sql("select *,'20231222' as etl_date,check01(get_time,used_time,pay_time) as max_time from t_df01 ")
    df03.show(5)
    df03.createOrReplaceTempView("t_df03")

    val df04=spark.sql("select * from t_df03 where max_time not in (select max_time from t_df02)")
    df04.show(5)

    spark.sql("set hive.exec.dynamic.partition.mode=nonstrict")
    df04.drop("max_time").write.insertInto("ods.coupon_use")

  }

}
快来参与讨论吧