在 Spark 中读取多行 CSV 文件

Question

我正在尝试在 spark 中读取多行 csv 文件。我的模式是：ID、名称和标记。 我的输入和实际输出如下所示。我没有得到预期的输出。有人可以帮助我在我的代码中缺少什么。

代码：

val myMarkDF =   spark
                .read
                .format("csv")
                .option("path","mypath\\marks.csv")
                .option("inferSchema","true")
                .option("multiLine","true")
                .option("delimiter",",")
                .load

输入：

1,A,
97,,
1,A,98
1,A,
99,,
2,B,100
2,B,95

实际输出：

+---+----+----+
|_c0| _c1| _c2|
+---+----+----+
|  1|   A|null|
| 97|null|null|
|  1|   A|  98|
|  1|   A|null|
| 99|null|null|
|  2|   B| 100|
|  2|   B|  95|
+---+----+----+

预期输出：

+---+----+----+
|_c0| _c1| _c2|
+---+----+----+
|  1|   A|  97|
|  1|   A|  98|
|  1|   A|  99|
|  2|   B| 100|
|  2|   B|  95|
+---+----+----+

谢谢！

Answer 1

编辑：一种更好的解决方案，可以处理更多类型的损坏记录（在第 2 列或第 3 列损坏）。重要的部分是计算非空条目的累积和，它将应该在同一记录中的行组合在一起。

val df = spark.read.csv("file.csv")
df.show
+---+----+----+
|_c0| _c1| _c2|
+---+----+----+
|  1|   A|null|
| 97|null|null|
|  1|   A|  98|
|  1|null|null|   <-- note that I intentionally changed these two rows
|  A|  99|null|   <-- to demonstrate how to handle two types of broken records
|  2|   B| 100|
|  2|   B|  95|
+---+----+----+

val df2 = df.withColumn(
    "id", monotonically_increasing_id()
).withColumn(
    "notnulls",
    $"_c0".isNotNull.cast("int") + $"_c1".isNotNull.cast("int") + $"_c2".isNotNull.cast("int")
).withColumn(
    "notnulls",
    ceil(sum($"notnulls").over(Window.orderBy("id")) / 3)
).groupBy("notnulls").agg(
    filter(
        flatten(collect_list(array("_c0","_c1","_c2"))),
        x => x.isNotNull
    ).alias("array")
).select(
    $"array"(0).alias("c0"),
    $"array"(1).alias("c1"),
    $"array"(2).alias("c2")
)

df2.show
+---+---+---+
| c0| c1| c2|
+---+---+---+
|  1|  A| 97|
|  1|  A| 98|
|  1|  A| 99|
|  2|  B|100|
|  2|  B| 95|
+---+---+---+

效果不佳的旧答案：

不是解析 csv 的最佳方式，但至少是您的用例的 MVP：

val df = sc.wholeTextFiles("marks.csv").map(
    row => row._2.replace(",,\n", "\n").replace(",\n", ",").split("\n")
).toDF(
    "value"
).select(
    explode($"value")
).select(
    split($"col", ",").as("col")
).select(
    $"col"(0), $"col"(1), $"col"(2)
)

df.show
+------+------+------+
|col[0]|col[1]|col[2]|
+------+------+------+
|     1|     A|    97|
|     1|     A|    98|
|     1|     A|    99|
|     2|     B|   100|
|     2|     B|    95|
+------+------+------+