python - 在 PySpark Streaming 中读取同一文件上具有不同架构的 CSV 文件

Question

我有一个 csv 文件，每行具有不同的长度，类似于:

left, 10, xdfe, 8992, 0.231
left, 10, xdfk, 8993, 2.231
right, 20, adfk, 8993, 2.231, DDT, 10, 10
right, 30, dfk, 923, 2.231, ADD, 10, 20
center, 923, 2.231, 10, 20
right, 34, efk, 326, 6.21, DDD, 20, 40

其中以关键字 left、right 和 center 开头的行具有相同的长度(left > 例如，行始终与其他 left 行具有相同的长度)。

我想使用 spark.readStream.csv 读取这些文件，进行一些取决于行类型的转换，并将结果写入 Parquet 。有没有办法根据每行第一列的值使用不同的架构？

Answer 1

不，您不能对同一个文件使用多个架构。您能做的最好的事情就是使用最长行的架构并将 mode 设置为 PERMISSIVE，这将为较短的行的缺失列提供 null 值。

不幸的是，这意味着如果缺少的列不在行末尾，则类型和列名称将会不同。例如。第三列是 right 行的字符串，可以是 center 行的 float (看起来应该是第五列)。一种方法是将所有内容读取为字符串，然后进行转换，但根据数据，某些列可以读取为例如 float 。

schema = StructType().add("a", "string").add("b", "string") \
    .add("c", "string").add("d", "string").add("e", "string") \
    .add("f", "string").add("g", "string").add("h", "string")

df = spark \
    .readStream \
    .option("mode", "PERMISSIVE") \
    .schema(schema) \
    .csv("/path/to/directory")

完成此操作后，可以对数据进行一些转换以获得外观正确的数据帧。下面的代码是 Scala 语言，但应该很容易转换为 python 并根据实际需要进行调整:

val df2 = df.select($"a", 
    when($"a" === "center", null).otherwise($"b").cast(FloatType).as("b"),
    when($"a" === "center", null).otherwise($"c").as("c"),
    when($"a" === "center", $"b").otherwise($"d").cast(FloatType).as("d"),
    when($"a" === "center", $"c").otherwise($"e").cast(FloatType).as("e"),
    $"f", $"g", $"h")

最终结果:

+------+----+-----+------+-----+----+----+----+
|     a|   b|    c|     d|    e|   f|   g|   h|
+------+----+-----+------+-----+----+----+----+
|  left|10.0| xdfe|8992.0|0.231|null|null|null|
|  left|10.0| xdfk|8993.0|2.231|null|null|null|
| right|20.0| adfk|8993.0|2.231| DDT|  10|  10|
| right|30.0|  dfk| 923.0|2.231| ADD|  10|  20|
|center|null| null| 923.0|2.231|null|null|null|
| right|34.0|  efk| 326.0| 6.21| DDD|  20|  40|
+------+----+-----+------+-----+----+----+----+