使用 Spark DataSet 聚合映射值的映射答案

【问题标题】：Aggregate Map of map values using Spark DataSet使用 Spark DataSet 聚合映射值的映射
【发布时间】：2017-09-01 15:07:42
【问题描述】：

我想使用 Spark DataSet 处理以下地图格式的 cassandra 列族。所以，我想将模型值分为两类溢价（City and Duster）与非溢价（Alto K10, Aspire, Nano and i10），我想要溢价与非溢价的最终计数为 2（City 和Duster 计数）与 10（Alto K10, Aspire, Nano and i10）。

代码：

  case class UserProfile(userdata:Map[String,Map[String,Int]])

 val userprofileDataSet = spark.read.format("org.apache.spark.sql.cassandra").options(Map("table"->"userprofilesagg","keyspace" -> "KEYSPACENAME")).load().as[UserProfile]

userprofileDataSet如何处理？？

数据格式：

 {'bodystyle': {'Compact Sedan': 1, 'Hatchback': 8, 'SUV': 1, 'Sedan': 4},   
    'models': {'Alto K10': 3, 'Aspire': 4, 'City': 1, 'Duster': 1, 'Nano': 3, 'i10': 2}}

已编辑的问题：

关于鱿鱼的回答。我现在想像这样汇总每个用户的结果：

    DOICvncGKUH9xBLnW3e9jXcd2 | non-premium | [Nano, Alto K10, Aspire, i10] | 12 | premium | [City, Duster] | 2
    BkkpgeAdCkYJEXsdZjiVz3bSb | non-premium | [Nano, Alto K10, Aspire, i10] | 17 | premium | [City, Duster] | 5

现在案例类看起来像这样

案例分类：

 case class UserProfile(userid:String, userdata:Map[String,Map[String,Int]])

数据：

DOICvncGKUH9xBLnW3e9jXcd2 | {'bodystyle': {'Compact Sedan': 1, 'Hatchback': 8, 'SUV': 1, 'Sedan': 4},   
    'models': {'Alto K10': 3, 'Aspire': 4, 'City': 1, 'Duster': 1, 'Nano': 3, 'i10': 2}}

BkkpgeAdCkYJEXsdZjiVz3bSb | {'bodystyle': {'Compact Sedan': 7, 'Hatchback': 5, 'SUV': 3, 'Sedan': 7},   
    'models': {'Alto K10': 1, 'Aspire': 7, 'City': 4, 'Duster': 1, 'Nano': 8, 'i10': 1}}

此外，您问我为什么提到 Bodystyle。这样我就可以将类似的聚合(SUV, Sedan) 应用为溢价并在其上保留非溢价。

【问题讨论】：

您只想处理模型吗？体型的作用是什么？
@squid 我已经编辑了这个问题。请你看看。

标签： apache-spark rdd apache-spark-dataset apache-spark-2.0

【解决方案1】：

我不确定bodystyle 的确切作用是什么。如果我正确理解了问题，那么您需要分类和计数，您可以尝试以下类似的操作，如果没有用，请删除 types：

--userprofile table
CREATE TABLE `userprofile`(
`properties` map<string,map<string,int>>);

--Aggregate by category
select category, 
       collect_set(type) as types, 
       sum(value) as count 
from (select case when lower(type) in ('city','duster') then 'premium'
             when lower(type) in ('alto k10', 'aspire', 'nano' , 'i10') then 'non-premium'
        end as category,
        type,value 
 from (select properties['models'] as models from userprofile) t
    lateral view explode(models) t as type, value)l group by category

输出

category    |   types                            |         count
non-premium | ["Aspire","i10","Nano","Alto K10"] |         12
premium     | ["City","Duster"]                  |         2

【讨论】：