用Scala将ods库中表数据全量抽取到Hive的dwd库中对应表中。表有timestamp类型， 均要求按照yyyy-MM-dd HH:mm:ss，不记录毫秒数，若原数据中只有年月日，则在时分秒的位置添加00:00:00，添加之后使其符合yyyy-MM-dd HH:mm:ss。 将ods库中customer表数据抽取到dwd库中dim_customer的分区表，分区字段为etldate且值与ods库的相对应表该值相等， 并添加dwd_insert_user、dwd_insert_time、dwd_modify_user、dwd_modify_time四列,其中dwd_insert_user、dwd_modify_user均填写“user1”， dwd_insert_time、dwd_modify_time均填写操作时间，并进行数据类型转换

以下是用Scala将ods库中表数据全量抽取到Hive的dwd库中对应表中的代码：

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("Extract Data from ODS to DWD")
  .enableHiveSupport()
  .getOrCreate()

val odsTable = "ods_table"
val dwdTable = "dwd_table"

val odsDF = spark.table(s"ods.$odsTable")

val dwdDF = odsDF
  .withColumn("timestamp_col", date_format($"timestamp_col", "yyyy-MM-dd HH:mm:ss"))
  .withColumn("timestamp_col", when(length($"timestamp_col") === 10, concat($"timestamp_col", lit(" 00:00:00")))
    .otherwise($"timestamp_col"))
  .select("col1", "col2", "timestamp_col")

dwdDF.write
  .mode("overwrite")
  .insertInto(s"dwd.$dwdTable")

以下是将ods库中customer表数据抽取到dwd库中dim_customer的分区表的代码：

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("Extract Data from ODS to DWD")
  .enableHiveSupport()
  .getOrCreate()

val odsTable = "customer"
val dwdTable = "dim_customer"

val etlDate = "2022-01-01" // replace with actual etl date

val odsDF = spark.table(s"ods.$odsTable")

val dwdDF = odsDF
  .withColumn("dwd_insert_user", lit("user1"))
  .withColumn("dwd_insert_time", current_timestamp())
  .withColumn("dwd_modify_user", lit("user1"))
  .withColumn("dwd_modify_time", current_timestamp())
  .withColumn("etldate", lit(etlDate))
  .select("col1", "col2", "etldate", "dwd_insert_user", "dwd_insert_time", "dwd_modify_user", "dwd_modify_time")

dwdDF.write
  .mode("overwrite")
  .partitionBy("etldate")
  .insertInto(s"dwd.$dwdTable")

以上代码中，我们使用了Spark SQL的API来处理数据，将timestamp类型的数据转换成指定格式，并添加了四列数据。最后，我们将数据写入到Hive的分区表中。