【问题标题】:Obtaining a monthly max value in PySpark RDD without SQL functions在没有 SQL 函数的情况下在 PySpark RDD 中获取每月最大值
【发布时间】:2020-06-18 00:30:48
【问题描述】:

我想找到这个查询的每月最大值,以获得每月最繁忙的机场。

在不使用 SQL 命令的情况下,是否有一个 pyspark 函数来获取最大值并将 RDD 减少到只有 12 行(每月 1 行)?

from pyspark import SparkContext
from pyspark.sql import SQLContext
from operator import add

sc = SparkContext.getOrCreate()
sqlContext = SQLContext(sc)

rows = sc.textFile("data.csv")
data = rows.map(lambda line: line.split(","))
header = data.first()

q = data.filter(lambda line:( line != header))\
    .map(lambda x: (x[1]+'-'+x[16], 1))\
    .reduceByKey(add)\
    .map(lambda x: (x[0].split('-'), x[1]))\
    .map(lambda x: (x[0][0], x[0][1], x[1]))\
    .collect()

sqlContext.createDataFrame(q, ['Month', 'Airport', 'Flights']).show(n=15)

+-----+-------+-------+
|Month|Airport|Flights|
+-----+-------+-------+
|    9|    DAL|   4388|
|    7|    STX|     43|
|    4|    PNS|    834|
|   12|    EUG|    549|
|    5|    ERI|    120|
|    9|    ATL|  34258|
|    7|    PHX|  17846|
|   12|    MYR|    434|
|   11|    CWA|    140|
|    9|    DAY|   1307|
|   12|    ADQ|     60|
|    1|    MTJ|    307|
|   10|    APF|      1|
|    2|    MAF|    657|
|    1|    SRQ|    687|
+-----+-------+-------+

【问题讨论】:

    标签: python apache-spark pyspark max rdd


    【解决方案1】:

    您可以尝试按键(月)对项目进行分组,以获取键值对 rdd,然后获取最大值,如下所示:

    max_by_month = (data
      .map(lambda x: (x[0], x))  # Convert to PairwiseRD by month
      # Take a maximum of the passed arguments by the last element (flights)
      .reduceByKey(lambda x1, x2: max(x1, x2, key=lambda x: x[-1])) 
      .values()) # Drop keys
    

    【讨论】:

    • 这是一个简单而优雅的解决方案,谢谢。注意:要使其与之前的代码一起运行,您首先必须删除:“.collect()”
    猜你喜欢
    • 2023-03-10
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2021-06-07
    • 2016-08-29
    相关资源
    最近更新 更多