时间戳

首页 > 解决方案 > Pyspark python reduceByKey 过滤器由 math.max

python - Pyspark python reduceByKey 过滤器由 math.max

问题描述

我有一个 rdd，其中 Key 为 date ， Value 为 . 就像是：

2017-07-03 00:00:00 (u'LANXESS', 17119)
2017-07-03 00:00:00 (u'SCHALTBAU', 225)
2018-05-08 00:00:00 (u'LINDE', 8105)
2018-05-08 00:00:00 (u'HSBC', 16084)

我想使用 rdd.reduceByKey 并且我想选择“somenumber”的最大值。

所以对于我希望看到的示例输入：

2017-07-03 00:00:00 (u'LANXESS', 17119)
2018-05-08 00:00:00 (u'HSBC', 16084)

这实际上是日期是我的关键，并且值被过滤了。

我正在尝试做类似的事情：

test4 = test3.reduceByKey(lambda a,b: math.max(a_2))

但我认为它是错误的。

我怎么能做到呢？

谢谢

标签： pythonapache-sparkfilterpyspark

解决方案

您可以使用带有 key 参数的 python 内置 max 函数。最大功能按键功能比较项目（在您的情况下为 lambda x: x[1] 或 itemgetter(1) ）。

rdd = sc.parallelize([
    ('2017-07-03 00:00:00', (u'LANXESS', 17119)),
    ('2017-07-03 00:00:00', (u'SCHALTBAU', 225)),
    ('2018-05-08 00:00:00', (u'LINDE', 8105)),
    ('2018-05-08 00:00:00', (u'HSBC', 16084)),
]).reduceByKey(lambda x,y: max((x, y), key=lambda x: x[1]))
rdd.collect()

或以更实用的方式

from operator import itemgetter
from functools import partial

reduce_func = partial(max, key=itemgetter(1))
rdd = sc.parallelize([
    ('2017-07-03 00:00:00', (u'LANXESS', 17119)),
    ('2017-07-03 00:00:00', (u'SCHALTBAU', 225)),
    ('2018-05-08 00:00:00', (u'LINDE', 8105)),
    ('2018-05-08 00:00:00', (u'HSBC', 16084)),
]).reduceByKey(reduce_func)
rdd.collect()

结果：

[('2018-05-08 00:00:00', ('HSBC', 16084)),
 ('2017-07-03 00:00:00', ('LANXESS', 17119))]

推荐阅读