Spark将文件中的键值对读取到数据帧中

Question

我需要读取日志文件并将其转换为spark数据帧

输入文件内容：

dateCreated   : 20200720
customerId    :  001
dateCreated   : 20200720
customerId    :  002
dateCreated   : 20200721
customerId    :  003

预期数据帧：

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

火花代码：

val spark = org.apache.spark.sql.SparkSession.builder.master("local").getOrCreate
    val inputFile = "C:\\log_data.txt"
    val rddFromFile = spark.sparkContext.textFile(inputFile)

    val rdd = rddFromFile.map(f => {
      f.split(":")
    })

    rdd.foreach(f => {
      println(f(0) + "\t" + f(1))
    })

如何将上述rdd转换为所需的DF

Answer 1

检查下面的代码

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       |
+     -+     +