检查以下代码。

scala> "cat /tmp/sample/input.csv".!
dateCreated   : 20200720
customerId    :  001
dateCreated   : 20200720
customerId    :  002
dateCreated   : 20200721
customerId    :  003

scala> val df = spark.read.text("/tmp/sample").select(split($"value",":").as("data"))
df: org.apache.spark.sql.DataFrame = [data: array<string>]

scala> df.show(false)
+---------------------------+
|data                       |
+---------------------------+
|[dateCreated   ,  20200720]|
|[customerId    ,   001]    |
|[dateCreated   ,  20200720]|
|[customerId    ,   002]    |
|[dateCreated   ,  20200721]|
|[customerId    ,   003]    |
+---------------------------+

scala> import org.apache.spark.sql.expressions._
import org.apache.spark.sql.expressions._

scala> val windowSpec = Window.orderBy($"id".asc)

scala> df
.select(trim($"data"(0)).as("data"),trim($"data"(1)).as("values"))
.select(map($"data",$"values").as("data"))
.select($"data"("dateCreated").as("dateCreated"),$"data"("customerId").as("customerId"))
.withColumn("id",monotonically_increasing_id)
.withColumn("customerId",lead($"customerId",1).over(windowSpec))
.where($"customerId".isNotNull)
.drop("id")
.show(false)

+-----------+----------+
|dateCreated|customerId|
+-----------+----------+
|20200720   |001       |
|20200720   |002       |
|20200721   |003       |
+-----------+----------+