如何在 Scala+Spark 中读取 csv 文件并将一列转换为 Map[String, String] 类型？

Question

我有一个包含几列的.csv 文件。

以一行为例：

aaa,bbb,{'foo': 'xxx', 'bar': 'zzz'}

我想阅读它并转换为以下类型的架构：

field1: String,
field2: String,
field3: Map[String, String]

我可以用这样的原始类型来做到这一点：

private val someSchema =
    StructType(
      StructField("field1", StringType, true) ::
      StructField("field2", StringType, true) ::
      StructField("field3", StringType, true) :: Nil)

     spark.read
      .format("csv")
      .option("header", true)
      .schema(someSchema)
      .load("path.csv")

但是当涉及到 Map[String, String] 时它不起作用，因为

线程“主”org.apache.spark.sql.AnalysisException 中的异常：CSV 数据源不支持 map 数据类型。

我该如何以另一种方式做到这一点？

Answer 1

您需要将其读取为字符串，然后使用 from_json 函数将其转换为 MapType 以将其解析为 map<string,string>：

val result = df.withColumn(
  "field3",
  from_json(col("field3"), lit("map<string,string>"))
)

---

但是，从您的示例来看，文件中的值似乎没有转义，spark 将无法解析它，因为您在列 field3 的值中有逗号（分隔符）。

在这种情况下，您可以将文件读取为文本，然后将 {} 中的逗号替换为另一个分隔符，例如 ;，并用 , 分割以获得 3 列并将列 field3 转换为映射使用str_to_map函数：

val df = spark.text("/path/file.csv")

val result = df.withColumn(
    "value",
    split(regexp_replace(col("value"), ",(?=[^{}]*\\})", ";"), ",")
).select(
    col("value")(0).as("field1"),
    col("value")(1).as("field2"),
    regexp_replace(col("value")(2), "[{}' ]", "").as("field3")
).withColumn(
    "field3",
    expr("str_to_map(field3, ';', ':')")
)

result.show
//+------+------+------------------------+
//|field1|field2|field3                  |
//+------+------+------------------------+
//|aaa   |bbb   |[foo -> xxx, bar -> zzz]|
//+------+------+------------------------+