在 Spark 中使用数据类型 map<string,bigint> 将数据帧写入 csv答案

【问题标题】：Write dataframe to csv with datatype map<string,bigint> in Spark在 Spark 中使用数据类型 map<string,bigint> 将数据帧写入 csv
【发布时间】：2018-08-12 01:50:01
【问题描述】：

我有一个文件 file1snappy.parquet。它有一个复杂的数据结构，比如地图，里面的数组。处理后我得到了最终结果。在将结果写入 csv 时我收到一些错误提示

"Exception in thread "main" java.lang.UnsupportedOperationException: CSV data source does not support map<string,bigint> data type."

我使用过的代码：

val conf=new SparkConf().setAppName("student-example").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlcontext = new org.apache.spark.sql.SQLContext(sc)
    val datadf = sqlcontext.read.parquet("C:\\file1.snappy.parquet")
    def sumaggr=udf((aggr: Map[String, collection.mutable.WrappedArray[Long]]) => if (aggr.keySet.contains("aggr")) aggr("aggr").sum else 0)
datadf.select(col("neid"),sumaggr(col("marks")).as("sum")).filter(col("sum") =!= 0).show(false)
    datadf.write.format("com.databricks.spark.csv").option("header", "true").save("C:\\myfile.csv")

我尝试转换 datadf.toString() 但我仍然面临同样的问题。如何将该结果写入 CSV。

火花版本 2.1.1

【问题讨论】：

标签： apache-spark apache-spark-sql rdd

【解决方案1】：

Spark CSV source 仅支持原子类型。你不能存储任何非原子的列

我认为最好的方法是为具有map<string,bigint> 作为数据类型的列创建一个 JSON，并将其保存在 csv 中，如下所示。

import spark.implicits._ 
import org.apache.spark.sql.functions._

datadf.withColumn("column_name_with_map_type", to_json(struct($"column_name_with_map_type"))).write.csv("outputpath")

希望这会有所帮助！

【讨论】：

它是说 $ 不是字符串上下文的成员
import spark.implicits._import org.apache.spark.sql.functions._ import这两个
在导入 spark.implicits._ 时表示找不到对象 spark
val spark = SparkSession.builder().appName("test").master("local[*]").getOrCreate()
如果你使用的是 spark2.0+，你应该使用SparkSession作为入口点，而不是SparkContext

【解决方案2】：

您正在尝试保存

的输出

val datadf = sqlcontext.read.parquet("C:\\file1.snappy.parquet")

我猜这是 udf 函数的错误，如果你这样做，所有的聚合都将徒劳无功

所以我想你想保存

的输出

datadf.select(col("neid"),sumaggr(col("marks")).as("sum")).filter(col("sum") =!= 0).show(false)

所以你需要将它保存在一个新的数据框变量中并使用该变量来保存。

val finalDF = datadf.select(col("neid"),sumaggr(col("marks")).as("sum")).filter(col("sum") =!= 0)
finalDF.write.format("com.databricks.spark.csv").option("header", "true").save("C:\\myfile.csv")

你应该没事的。

【讨论】：