3

我正在尝试使用 apache spark (pyspark) 对值 (key, value) 进行分组。我设法通过键进行分组，但在内部我想对值进行分组，如下例所示。

我需要按 GYEAR 列按 cout() 分组。

%pyspark

rdd1 = sc.textFile("/datos/apat63_99.txt")

rdd2 = rdd1.map(lambda line :  line.split(",") ).map(lambda l : (l[4],l[1],l[0]))

for line in rdd2.take(6):
    print(line)

######################

rdd3 = rdd2.map(lambda line:(line[0],(line[1:]) ))

rddx = rdd2.groupByKey()
rddx.take(5)

我希望输出是：

在：

(u'"COUNTRY"', u'"GYEAR"', u'"PATENT"')
(u'"BE"', u'1963', u'3070801')
(u'"BE"', u'1964', u'3070811')
(u'"US"', u'1963', u'3070802')
(u'"US"', u'1963', u'3070803')
(u'"US"', u'1963', u'3070804')
(u'"US"', u'1963', u'3070805')
(u'"US"', u'1964', u'3070807')

出去：

(u'"BE"', [(u'1963', 1), (u'1964', 1)])
(u'"US"', [(u'1963', 4), (u'1964', 2)])
4

3 回答 3

5

这是使用 RDD 方法的一种方法：

from operator import add

# initialize the RDD
rdd = sc.parallelize([(u'"COUNTRY"', u'"GYEAR"', u'"PATENT"')
    , (u'"BE"', u'1963', u'3070801')
    , (u'"BE"', u'1964', u'3070811')
    , (u'"US"', u'1963', u'3070802')
    , (u'"US"', u'1963', u'3070803')
    , (u'"US"', u'1963', u'3070804')
    , (u'"US"', u'1963', u'3070805')
    , (u'"US"', u'1964', u'3070807')])

请执行下列操作：

  1. 将元组设置(COUNTRY, GYEAR)为键，1值
  2. 用 reduceByKey(add) 计算键
  3. 调整 key 到COUNTRY, value 到[(GYEAR, cnt)]cnt是从之前的 reduceByKey 计算出来的
  4. 运行reduceByKey(add)以合并具有相同键（COUNTRY）的列表。

  5. 使用过滤器删除标题

    rdd_new = rdd.map(lambda x: ((x[0],x[1]), 1) ) \
             .reduceByKey(add) \
             .map(lambda x: (x[0][0], [(x[0][1],x[1])])) \
             .reduceByKey(add) \
             .filter(lambda x: x[0] != '"COUNTRY"')

检查结果：

>>> rdd_new.take(2)
[(u'"US"', [(u'1964', 1), (u'1963', 4)]),
 (u'"BE"', [(u'1963', 1), (u'1964', 1)])]
于 2019-07-06T02:36:45.953 回答
2

正如@PIG 所指出的，使用 DataFrame 比使用 RDD 更容易。

另外，我建议使用create_map,collect_list和你自己的 UDF 到combine_maps. 这应该允许您继续使用结构化数据。

df2=df1.withColumn('result',F.create_map(df1.Gyear, df1.Patient))
df2.show()

+-------+-----+-------+-----------+
|country|Gyear|Patient|     result|
+-------+-----+-------+-----------+
|      S| 1963|      4|[1963 -> 4]|
|     BE| 1963|      1|[1963 -> 1]|
|      S| 1964|      1|[1964 -> 1]|
|     BE| 1964|      1|[1964 -> 1]|
+-------+-----+-------+-----------+

from typing import List, Dict
from pyspark.sql.functions import udf
from functools import reduce
from pyspark.sql.types import *

def combine_map(x: Dict[str, int], y: Dict[str, int]) -> Dict[str, int]:
    return {k: x.get(k, 0) + y.get(k, 0) for k in set(x) | set(y)}


@udf(returnType=MapType(StringType(), IntegerType()))
def combine_maps(maps):
    return reduce(combine_map, maps, {})

df2.groupBy("country").agg(F.collect_list("result").alias("result")) \
   .withColumn("result", combine_maps("result")) \
   .show(truncate=False) 

+-------+----------------------+
|country|result                |
+-------+----------------------+
|S      |[1964 -> 1, 1963 -> 4]|
|BE     |[1964 -> 1, 1963 -> 1]|
+-------+----------------------+
于 2019-07-05T10:40:38.800 回答
1

是您要找的吗？。我无法将结果列转换为元组，因此只需将其连接为字符串。以下解决方案的性能可能不佳。

在火花 2.3 Ubuntu 18.04 上运行

    from pyspark.sql import SparkSession
    import pyspark.sql.functions as F

    spark = SparkSession.builder.appName("test").enableHiveSupport().getOrCreate()


    l = [
    ('BE', '1963', '3070801'),
    ('BE', '1964', '3070811'),
    ('S', '1963', '3070802'),
    ('S', '1963', '3070803'),
    ('S', '1963', '3070804'),
    ('S', '1963', '3070805'),
    ('S', '1964', '3070807')]

    colmns = ['country', 'Gyear', 'Patient']

    df=spark.createDataFrame(l, colmns)
    df.show()


+-------+-----+-------+
|country|Gyear|Patient|
+-------+-----+-------+
|     BE| 1963|3070801|
|     BE| 1964|3070811|
|      S| 1963|3070802|
|      S| 1963|3070803|
|      S| 1963|3070804|
|      S| 1963|3070805|
|      S| 1964|3070807|
+-------+-----+-------+

    df1=df.groupBy("country","Gyear").agg(F.count("Patient").alias("Patient"))
    df1.show()

+-------+-----+-------+
|country|Gyear|Patient|
+-------+-----+-------+
|      S| 1963|      4|
|     BE| 1963|      1|
|      S| 1964|      1|
|     BE| 1964|      1|
+-------+-----+-------+
    df2=df1.withColumn('result',F.concat(F.lit('('),df1.Gyear,F.lit(','),df1.Patient,F.lit(')'))).drop("Gyear","Patient")
    df2.show()

+-------+--------+
|country|  result|
+-------+--------+
|      S|(1963,4)|
|     BE|(1963,1)|
|      S|(1964,1)|
|     BE|(1964,1)|
+-------+--------+    

    df2.groupBy("country").agg(F.collect_list("result")).show()

+-------+--------------------+
|country|collect_list(result)|
+-------+--------------------+
|      S|[(1963,4), (1964,1)]|
|     BE|[(1963,1), (1964,1)]|
+-------+--------------------+
于 2019-07-05T07:06:16.293 回答