Filtrage de données dans un CA

J'ai réussi à pré traiter mes données dans pyspark pour obtenir quelque chose comme ceci

[(u'key1', u'1'), (u'key2', u'1'), (u'key1', u'2'), (u'key3', u'2'), (u'key4', u'1'), (u'key1', u'4'), (u'key5', u'1'), (u'key6', u'2'), (u'key7', u'4'), (u'key8', u'5'), (u'key9', u'6'), (u'key10', u'7')]

Maintenant, j'ai besoin de filtrer sur la base de ces conditions :

1) les valeurs de filtre associé à au moins 2 clés.

sortie uniquement ceux (k,v) paires qui a '1','2','4' les valeurs doivent être présents, car ils sont associés à plus de 2 touches

 [(u'key1', u'1'), (u'key2', u'1'), (u'key1', u'2'), (u'key3', u'2'), (u'key4', u'1'), (u'key1', u'4'), (u'key5', u'1'), (u'key6', u'2'), (u'key2', u'4')]

2) filtre clés qui sont associés à au moins 2 valeurs

sortie uniquement ceux (k,v) paires qui a key1, key2 que les clés doivent être là car ils sont associés à au moins 2 valeurs

[(u'key1', u'1'), (u'key2', u'1'), (u'key1', u'2'), (u'key1', u'4'), (u'key2', u'4')]

Toute suggestion serait d'une grande aide.

Mise à jour : j'ai utilisé groupBy et un filtre de groupe pour les touches avec plusieurs valeurs

 [(u'key1', [u'1', u'2', u'4']), (u'key2',[u'1', u'4'])]

Maintenant, comment puis-je partager ce (clé, (liste de valeurs)) à chaque (k,v) paire d'appliquer des transformations à venir ?

Vous pouvez faire tout en un seul passage - reduceByKey, filtrer les éléments qui ont plus que 2 valeurs, puis recueillir ou de processus de ce qui est là. La partie spécifiquement vous rencontrez des problèmes avec?
Réduire de clé d'agrégation basé sur la touche de droite ? donc ça donnerait quelque chose comme (u'key1', u'1,2,3') si je rentre les valeurs dans reduceByKey séparés par des ','. Je n'ai pas besoin de mes données pour obtenir l'agrégation. s'il vous plaît corrigez-moi si je me trompe.
Si j'utilise reduceByKey et puis une fois qu'il est agrégé je peux utiliser un filtre à seulement ceux qui dispose de plus de 2 valeurs. Maintenant, comment dois-je effectuer second tour de filtrage pour filtrer les clés qui sont associés à plus de 2 valeurs ?
une fois que vous avez des groupes comme key1 -> 1, 2, 3 vous pouvez filtrer en fonction de la taille des valeurs (len >= 2) et de recueillir à la fois des clés et des valeurs.

InformationsquelleAutor Magic | 2016-11-14

my_rdd = sc.parallelize([(u'key1', u'1'), (u'key2', u'1'), (u'key1', u'2'), (u'key2', u'3'), (u'key4', u'1'), (u'key1', u'4'), (u'key4', u'1'), (u'key6', u'2'), (u'key7', u'4'), (u'key8', u'5'), (u'key9', u'6'), (u'key10', u'7')])

#filter keys which are associated to atleast 2 values

filter2_rdd = my_rdd.groupByKey() \
                    .mapValues(lambda x: list(x)) \
                    .filter(lambda x: len(x[1])>=2) \
                    .flatMap(lambda x: [(x[0],item) for item in x[1]])

#filter values associated to atleast 2 keys.
filte1_rdd = filter2_rdd.map(lambda x: (x[1],x[0])) \
                        .groupByKey().mapValues(lambda x: list(x))\
                        .filter(lambda x: len(x[1])>=2)\
                        .flatMap(lambda x: [(item,x[0]) for item in x[1]])

Cela va fonctionner!!

A fait presque toutes les choses, à l'exception de la flatMap partie où vous divisez la liste en arrière individuels (k,v) paire. FlatMap partie aidé, Merci!

InformationsquelleAutor sau

Réduire par la clé, filtre et de le rejoindre:

>>> rdd.mapValues(lambda _: 1) \  # Add key of value 1
...     .reduceByKey(lambda x, y: x + y) \ # Count keys
...     .filter(lambda x: x[1] >= 2) \ # Keep only if number is >= 2
...     .join(rdd) # join with original (serves as filter)
...     .mapValues(lambda x: x[0]) # reshape

pouvez-vous nous expliquer ce que vous essayez de faire ?

InformationsquelleAutor

Vous devez vous connecter pour publier un commentaire.