代码之家 › 专栏 › 技术社区 › Kalpesh

如何将GroupedDataset保存到parquet或将其转换为toDF

apache-spark-dataset group-by apache-spark scala

Kalpesh · 技术社区 · 7 年前

我正在使用spark 1.6.1。

是否有任何API可用于将GroupDataset保存到拼花文件。

E、我有一个自定义对象“Procedure”,我已经将数据帧转换为Procedure对象。之后,我在patientID上进行分组。我想将DDS分组到拼花文件中,或者将其作为数据帧传递给其他函数。我没有为存储获取任何API,也没有将其转换为数据帧。

val procedureDs: Dataset[Procedure] = joinDf.select("patientid", "patientprocedureid", "procedurecode").as[Procedure]
val groupedDs:GroupedDataset[Long, Procedure] = procedureDs.groupBy{ x => x.patientid }

应用地图组后

val a = groupedDs.mapGroups{ case (k,vs) => { (k, vs.toSeq)}}

它给出了以下错误:

Exception in thread "main" java.lang.UnsupportedOperationException: No Encoder found for com.....PatientDiagnosis
- array element class: "com....PatientDiagnosis"
- field (class: "scala.collection.Seq", name: "_2")
- root class: "scala.Tuple2"

我试图给出明确的编码器

val a = groupedDigDs.mapGroups((k,vs) =>  (k, vs.toSeq))(org.apache.spark.sql.Encoders.bean(classOf[(Long, Seq[com....PatientDiagnosis])]))

然后错误更改为:

 java.lang.UnsupportedOperationException: Cannot infer type for class scala.Tuple2 because it is not bean-compliant

1 回复 | 直到 7 年前

zero323 little_kid_pea 7 年前

等同于 GroupedData ( RelationalGroupedDataset 在Spark 2中。x) , GroupedDataset ( KeyValueGroupedDataset 在Spark 2中。x) 必须在保存之前进行聚合。

如果你的目标是另一个 groupByKey 您可以使用 mapGroups :

val groupedDs: GroupedDataset[K, V] = ???
// ... { case (k, xs) => (k, xs.toSeq) }  to preserve key as well
groupedDs.mapGroups { case (_, xs) => xs.toSeq }

推荐文章

John Smith · Java按值对映射进行分组,其中值为列表

2 年前

The Great · 拆分并存储数据帧,但名称基于特定列中的唯一值

2 年前

Programming Noob · AttributeError:“SeriesGroupBy”对象没有属性“tolist”

2 年前

Flo · 分组依据中的SQL大小写

2 年前

user13663655 · 我可以使用条件分组并返回源类型值吗

2 年前

pinkiBet · 使用MAX和UNION分组,还是加入?

2 年前

Shubham · 如何在R中查找数据表的两个多类别列之间的差异/setdiff()

2 年前

user7052482 · 将来自联合的SQL结果分组(SQL Server)

6 年前

lugger1 · 如何获取Postgres中时间间隔的平均值

6 年前

JustCallMeGary · 如何使用R中的一个因子操作两个变量[重复]

6 年前