com.datastax.spark.connector.types.TypeConversionException：无法将 scala.collection.immutable.Map$Map1 类型的对象映射转换为（AnyRef，AnyRef）答案

【问题标题】：com.datastax.spark.connector.types.TypeConversionException: Cannot convert object Map of type scala.collection.immutable.Map$Map1 to (AnyRef, AnyRef)com.datastax.spark.connector.types.TypeConversionException：无法将 scala.collection.immutable.Map$Map1 类型的对象映射转换为（AnyRef，AnyRef）
【发布时间】：2021-10-24 11:31:13
【问题描述】：

我有一个以下架构的 Spark DataFrame。

root
 |-- partition_key: string (nullable = true)
 |-- row_key: string (nullable = true)
 |-- attributes: map (nullable = true)
 |    |-- key: string
 |    |-- value: string (valueContainsNull = true)
 |-- data_as_of_date: string (nullable = true)

我正在向 Cassandra 表写入相同的内容。

Cassandra 表架构如下：

create table provision_bmss.bmss_cust (
  partition_key text,
  row_key text,
  group int,
  attributes map<text,text>,
  data_as_of_date text,
  PRIMARY KEY (partition_key, row_key, group)
  )
  WITH cdc = 'FALSE'
  AND default_time_to_live = '34560000';

我正在使用 Spark Datastax 连接器在以下逻辑之后写入表：

val maxItem = 65000
  
  dataFrame.select($"partition_key", $"row_key", $"data_as_of_date", posexplode($"attributes"))
  .withColumn("group", $"pos".divide(maxItem).cast("int"))
  .groupBy($"partition_key", $"row_key", $"data_as_of_date", $"group")
  .agg(collect_list(map($"key", $"value")).as("attributes"))
  .select($"partition_key", $"row_key", $"group", $"attributes", $"data_as_of_date")
  .write
  .format("org.apache.spark.sql.cassandra")
  .mode("append")
  .options(Map( "keyspace" -> keySpace, "table" -> tableName ))
  .save()

我收到以下错误：

com.datastax.spark.connector.types.TypeConversionException: Cannot convert object Map(cli_rel_typ_c_00001 -> 01) of type class scala.collection.immutable.Map$Map1 to (AnyRef, AnyRef)

我认为这与代码中的.agg(collect_list(map($"key", $"value")).as("attributes")) 行有关。

这里，Map 中的所有内容都是 <String, String> 类型

我无法解决同样的问题。有人可以帮忙吗？

输出的DataFrame架构如下（与预期不符）：

root
 |-- partition_key: string (nullable = true)
 |-- row_key: string (nullable = true)
 |-- group: int (nullable = true)
 |-- attributes: array (nullable = true)
 |    |-- element: map (containsNull = true)
 |    |    |-- key: string
 |    |    |-- value: string (valueContainsNull = true)
 |-- data_as_of_date: string (nullable = true)

预期的输出 DataFrame 架构如下：

root
     |-- partition_key: string (nullable = true)
     |-- row_key: string (nullable = true)
     |-- group: int (nullable = true)
     |-- attributes: map (nullable = true)
     |    |-- key: string
     |    |-- value: string (valueContainsNull = true)
     |-- data_as_of_date: string (nullable = true)

这个问题的前置问题是Divide Spark DataFrame rows into multiple rows depending on the size of Map type column

【问题讨论】：

Cassandra 中表的架构是什么？
@Alex Ott - 添加了相关表的架构。

标签： scala apache-spark cassandra datastax

【解决方案1】：

我推荐了这篇文章 - How to use groupBy to collect rows into a map?

我可以使用 flatten 和 toMap 函数来做同样的事情。

更新的工作代码如下：

val joinMap = udf {
  values: Seq[Map[String,String]] => values.flatten.toMap
}

def writeToCassandra(dataFrame: DataFrame, keySpace: String, tableName: String) = {
   
  val maxItem = 65000
  
  val dfPreFinal = 
  dataFrame.select($"partition_key", $"row_key", $"data_as_of_date", posexplode($"attributes"))
  .withColumn("group", $"pos".divide(maxItem).cast("int"))
  .groupBy($"partition_key", $"row_key", $"data_as_of_date", $"group")
  .agg(collect_list(map($"key", $"value")).as("attributes"))
  .select($"partition_key", $"row_key", $"group", $"attributes", $"data_as_of_date")
  
  val dfFinal = dfPreFinal.withColumn("attributes", joinMap(col("attributes")))
  
  .write
  .format("org.apache.spark.sql.cassandra")
  .mode("append")
  .options(Map( "keyspace" -> keySpace, "table" -> tableName ))
  .save() 
}

有没有更好的方法（可能没有UDF）？

【讨论】：