pyspark：grouby然后获得每组的最大值

2 回答

• 11

  这里没有 groupBy . 简单 reduceByKey 会很好，大部分时间会更有效：

  data_file = sc.parallelize([
     (u'u1', u's1', 20), (u'u1', u's2', 5),
     (u'u2', u's3', 5), (u'u2', u's2', 10)])
  
  max_by_group = (data_file
    .map(lambda x: (x[0], x))  # Convert to PairwiseRD
    # Take maximum of the passed arguments by the last element (key)
    # equivalent to:
    # lambda x, y: x if x[-1] > y[-1] else y
    .reduceByKey(lambda x1, x2: max(x1, x2, key=lambda x: x[-1])) 
    .values()) # Drop keys
  
  max_by_group.collect()
  ## [('u2', 's2', 10), ('u1', 's1', 20)]
  

  回复于 2024-05-14T09:41:58+08:00
• 2

  我想我找到了解决方案：

  from pyspark import SparkContext, SparkConf
  
  def reduce_by_max(rdd):
      """
      Helper function to find the max value in a list of values i.e. triplets. 
      """
      max_val = rdd[0][2]
      the_index = 0
  
      for idx, val in enumerate(rdd):
          if val[2] > max_val:
              max_val = val[2]
              the_index = idx
  
      return rdd[the_index]
  
  conf = SparkConf() \
      .setAppName("Collaborative Filter") \
      .set("spark.executor.memory", "5g")
  sc = SparkContext(conf=conf)
  
  # some file contains tuples ('user', 'item', 'occurrences')
  data_file = sc.textData('file:///some_file.txt')
  
  # Create the triplet so I can index stuff
  data_file = data_file.map(lambda l: l.split()).map(lambda l: (l[0], l[1], float(l[2])))
  
  # Group by the user i.e. r[0]
  grouped = data_file.groupBy(lambda r: r[0])
  
  # Get the values as a list
  group_list = grouped.map(lambda x: (list(x[1]))) 
  
  # Get the max value for each user. 
  max_list = group_list.map(reduce_by_max).collect()
  

  回复于 2024-05-14T09:41:58+08:00