相关疑难解决方法(0)

我有一个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)
• 来自小时组== …

目标和动机

该(x, y)API已经日益普及,多年来,然而,没有关于它的一切是完全理解在结构方面,工作和相关的操作.

一个重要的操作是过滤.过滤是一种常见的要求,但用例是多种多样的.因此,某些方法和功能将比其他用例更适用于某些用例.

总之,本文的目的是触及一些常见的过滤问题和用例,演示解决这些问题的各种不同方法,并讨论它们的适用性.本文试图解决的一些高级问题是

• 基于单个值/标签切片
• 基于来自一个或多个级别的多个标签进行切片
• 过滤布尔条件和表达式
• 哪种方法适用于什么情况

这些问题已分解为6个具体问题,如下所列.为简单起见,以下设置中的示例DataFrame仅具有两个级别,并且没有重复的索引键.提出问题的大多数解决方案可以推广到N级.

本文不会介绍如何创建MultiIndexes,如何对它们执行赋值操作,或任何与性能相关的讨论(这些是另一个时间的单独主题).

问题

问题1-6将在上下文中询问下面的设置.

mux = pd.MultiIndex.from_arrays([
    list('aaaabbbbbccddddd'),
    list('tuvwtuvwtuvwtuvw')
], names=['one', 'two'])

df = pd.DataFrame({'col': np.arange(len(mux))}, mux)

         col
one two     
a   t      0
    u      1
    v      2
    w      3
b   t      4
    u      5
    v      6
    w      7
    t      8
c   u      9
    v     10
d   w     11
    t     12
    u     13
    v     14
    w     15

Run Code Online (Sandbox Code Playgroud)

问题1:选择单个项目
如何在"1"级中选择"a"的行？

         col
one two     
a   t      0 …

我花了相当多的时间阅读pyspark和spark- dataframe标签的一些问题,而且我经常发现海报没有提供足够的信息来真正理解他们的问题.我经常评论要求他们发布MCVE,但有时让他们显示一些样本输入/输出数据就像拔牙一样.例如:请参阅有关此问题的评论.

也许问题的一部分是人们只是不知道如何轻松地为火花数据帧创建MCVE.我认为将这个pandas问题的spark-dataframe版本作为可以链接的指南是有用的.

那么如何创造一个好的,可重复的例子呢？

我正在研究一个代表事件流的数据集(比如从网站上发布的跟踪事件).所有活动都有时间戳.我们经常遇到的一个用例是尝试找到给定字段的第一个非空值.因此,举例来说,最让我们感受到的是:

val eventsDf = spark.read.json(jsonEventsPath) 

case class ProjectedFields(visitId: String, userId: Int, timestamp: Long ... )

val projectedEventsDs = eventsDf.select(
    eventsDf("message.visit.id").alias("visitId"),
    eventsDf("message.property.user_id").alias("userId"),
    eventsDf("message.property.timestamp"),

    ...

).as[ProjectedFields]

projectedEventsDs.groupBy($"visitId").agg(first($"userId", true))

上述代码的问题first在于无法保证馈送到该聚合函数的数据的顺序.我希望它被排序timestamp以确保它是时间戳的第一个非null userId而不是任何随机的非null userId.

有没有办法在分组中定义排序？

使用Spark 2.10

BTW,在SPARK DataFrame中为Spark 2.10建议的方式:选择每个组的第一行是在分组之前进行排序 - 这不起作用.例如,以下代码:

case class OrderedKeyValue(key: String, value: String, ordering: Int)
val ds = Seq(
  OrderedKeyValue("a", null, 1), 
  OrderedKeyValue("a", null, 2), 
  OrderedKeyValue("a", "x", 3), 
  OrderedKeyValue("a", "y", 4), 
  OrderedKeyValue("a", null, 5)
).toDS()

ds.orderBy("ordering").groupBy("key").agg(first("value", true)).collect()

有时会返回Array([a,y]),有时Array([a,x])