pyspark: grouby et puis obtenir la valeur maximum de chaque groupe

J'aimerais groupe par une valeur et ensuite trouver la valeur max dans chaque groupe à l'aide de PySpark. J'ai le code suivant mais maintenant, je suis un peu coincé sur la façon d'extraire la valeur de max. 

# some file contains tuples ('user', 'item', 'occurrences')
data_file = sc.textData('file:///some_file.txt')
# Create the triplet so I 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])
# Here is where I am stuck 
group_list = grouped.map(lambda x: (list(x[1]))) #?

Retourne quelque chose comme:

[[(u'u1', u's1', 20), (u'u1', u's2', 5)], [(u'u2', u's3', 5), (u'u2', u's2', 10)]]

Je veux trouver un max 'accident" pour chaque utilisateur. Le résultat final après avoir fait le max entraînerait une EDR qui ressemblait à ceci: 

[[(u'u1', u's1', 20)], [(u'u2', u's2', 10)]]

Où seul le max dataset restera pour chacun des utilisateurs dans le fichier. En d'autres termes, je veux changer le valeur de la RDD ne contiennent qu'un seul triplet chaque utilisateurs max occurrences.

InformationsquelleAutor user985030 | 2015-11-15
  1. 12

    Il n'est pas nécessaire pour groupBy ici. Simple reduceByKey ferait l'affaire, et la plupart du temps sera plus efficace:

    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)]
    • pourriez-vous expliquer ce (lambda x1, x2: max(x1, x2, key=lambda x: x[-1])) si possible?
    • est juste un standard de Python max. Il faut éléments et renvoie le plus grand. key argument décrit la façon dont les éléments doivent être comparées (ici par le dernier élément).
    InformationsquelleAutor zero323
  2. 2

    Je pense que j'ai trouvé la solution: 

    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()
    InformationsquelleAutor user985030