小编Ram*_*ami的帖子

我有一个DataFrame生成如下:

df.groupBy($"Hour", $"Category")
  .agg(sum($"value") as "TotalValue")
  .sort($"Hour".asc, $"TotalValue".desc))

结果如下:

+----+--------+----------+
|Hour|Category|TotalValue|
+----+--------+----------+
|   0|   cat26|      30.9|
|   0|   cat13|      22.1|
|   0|   cat95|      19.6|
|   0|  cat105|       1.3|
|   1|   cat67|      28.5|
|   1|    cat4|      26.8|
|   1|   cat13|      12.6|
|   1|   cat23|       5.3|
|   2|   cat56|      39.6|
|   2|   cat40|      29.7|
|   2|  cat187|      27.9|
|   2|   cat68|       9.8|
|   3|    cat8|      35.6|
| ...|    ....|      ....|
+----+--------+----------+

如您所见,DataFrame按Hour递增顺序排序,然后按TotalValue降序排序.

我想选择每组的顶行,即

• 来自小时组== 0选择(0,cat26,30.9)
• 来自小时组== 1选择(1,cat67,28.5)
• 来自小时组== …

我正在尝试为每个用户构建一个包含每小时每小时平均记录数的向量.因此,矢量必须具有24维.

我的原始DataFrame有userID和hour列,我开始做一个groupBy并计算每个用户每小时的记录数,如下所示:

val hourFreqDF = df.groupBy("userID", "hour").agg(count("*") as "hfreq")

现在,为了根据本答案中的第一个建议,我按照每个用户生成一个向量.

val hours = (0 to 23 map { n => s"$n" } toArray)

val assembler = new VectorAssembler()
                     .setInputCols(hours)
                     .setOutputCol("hourlyConnections")

val exprs = hours.map(c => avg(when($"hour" === c, $"hfreq").otherwise(lit(0))).alias(c))

val transformed = assembler.transform(hourFreqDF.groupBy($"userID")
                           .agg(exprs.head, exprs.tail: _*))

当我运行此示例时,我收到以下警告:

Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.debug.maxToStringFields' in SparkEnv.conf.

我认为这是因为表达太长了？

我的问题是:我能安全地忽略这个警告吗？

我有一个两列的DataFrame,ID类型Int和Vec类型Vector(org.apache.spark.mllib.linalg.Vector).

DataFrame如下所示:

ID,Vec
1,[0,0,5]
1,[4,0,1]
1,[1,2,1]
2,[7,5,0]
2,[3,3,4]
3,[0,8,1]
3,[0,0,1]
3,[7,7,7]
....

我想groupBy($"ID")通过对向量求和来对每个组内的行应用聚合.

上述示例的所需输出将是:

ID,SumOfVectors
1,[5,2,7]
2,[10,8,4]
3,[7,15,9]
...

可用的聚合函数将不起作用,例如,df.groupBy($"ID").agg(sum($"Vec")将导致ClassCastException.

如何实现自定义聚合函数,允许我进行向量或数组的总和或任何其他自定义操作？

问题几乎在标题中:是否有一种有效的方法来计算DataFrame中每列中的不同值？

该描述方法只提供了计数,但不重复计数,我不知道是否有AA的方式来获得所有的重复计数(或选择)列.

我有一个DataFrame看起来如下:

userID, category, frequency
1,cat1,1
1,cat2,3
1,cat9,5
2,cat4,6
2,cat9,2
2,cat10,1
3,cat1,5
3,cat7,16
3,cat8,2

不同类别的数量是10,我想为每个userID类别创建一个特征向量,并用零填充缺少的类别.

所以输出将是这样的:

userID,feature
1,[1,3,0,0,0,0,0,0,5,0]
2,[0,0,0,6,0,0,0,0,2,1]
3,[5,0,0,0,0,0,16,2,0,0]

这只是一个说明性的例子,实际上我有大约200,000个唯一的userID和300个独特的类别.

创建功能的最有效方法是什么DataFrame？

我的目标是构建一个多字符分类器.

我已经构建了一个用于特征提取的管道,它包括一个StringIndexer转换器,用于将每个类名映射到一个标签,该标签将用于分类器训练步骤.

管道安装在训练集上.

测试集必须由拟合的管道处理,以便提取相同的特征向量.

知道我的测试集文件具有与训练集相同的结构.这里可能的情况是在测试集中面对一个看不见的类名,在这种情况下,StringIndexer将无法找到标签,并且将引发异常.

这种情况有解决方案吗？或者我们如何避免这种情况发生？

我试图通过仅保留那些具有某个字符串列非空的行来过滤DataFrame.

操作如下:

df.filter($"stringColumn" !== "")

我的编译器显示,自从我转到Spark 2.0.1后,不推荐使用!==

如何在Spark> 2.0中检查字符串列值是否为空？

我有兴趣使用Spark MLlib应用LDA主题建模.我已经检查了这里的代码和解释,但我找不到如何使用模型然后在一个新的看不见的文档中找到主题分布.

阅读完这个问题后,我想提出更多问题:

1. Cluster Manager是一个长期运行的服务,它在哪个节点上运行？
2. Master和Driver节点是否可能是同一台机器？我认为应该有一个规则说明这两个节点应该是不同的？
3. 如果Driver节点出现故障,谁负责重新启动应用程序？什么会发生什么？即主节点,Cluster Manager和Workers节点将如何参与(如果它们)以及以何种顺序？
4. 与上一个问题类似:如果主节点出现故障,将会发生什么,以及谁负责从故障中恢复？

我希望(在我的Android应用程序中使用Google maps api v2)根据类别隐藏或显示我的GoogleMap对象上的标记,就像在google maps web api中一样,例如:

我有一个包含50个标记的GoogleMap,其中20个代表餐馆,20个代表公交车站,10个是电影院.

如果我们取消勾选一个复选框,可以在Android google maps api v2上对这些标记进行过滤吗？

我想做类似的事情,但在我的Android设备上使用谷歌地图api v2:http://www.geocodezip.com/v3_MW_example_categories.html

对不起基本问题,但我是初学者.