Замена groupBykey () с помощью reduceByKey () [duplicate]

Я немного опаздываю на разговор, но вот мое предложение:

>>> foo = sc.parallelize([(1, ('a','b')), (2, ('c','d')), (1, ('x','y'))])
>>> foo.map(lambda (x,y): (x, [y])).reduceByKey(lambda p,q: p+q).collect()
[(1, [('a', 'b'), ('x', 'y')]), (2, [('c', 'd')])]

Сообщение об ошибке связано с типом «a» в вашем закрытии.

 My_KMV = My_KV.reduce(lambda a, b: a.append([b]))

Пусть pySpark явно оценивает a как список. Например,

My_KMV = My_KV.reduceByKey(lambda a,b:[a].extend([b]))

Во многих случаях reduceByKey будет предпочтительнее groupByKey, обратитесь к: http://databricks.gitbooks.io/databricks-spark-knowledge-base/content/ best_practices / prefer_reducebykey_over_groupbykey.html

Хорошо. Надеюсь, у меня все получилось. Ваш вход выглядит примерно так:

kv_input = [("a", 1), ("a", 2), ("a", 3), ("b", 1), ("b", 5)]

, и вы хотите получить что-то вроде этого:

kmv_output = [("a", [1, 2, 3]), ("b", [1, 5])]

Тогда это может выполнить задание (см. здесь ):

d = dict()
for k, v in kv_input:
    d.setdefault(k, list()).append(v)
kmv_output = list(d.items())

Если я ошибаюсь, скажите, пожалуйста, чтобы я мог настроить это на ваши нужды.

PS: a.append([b]) всегда возвращается None , Возможно, вы захотите наблюдать либо [b], либо a, но не результат append.

Я попытался с помощью combByKey, вот мои шаги

combineddatardd=sc.parallelize([("A", 3), ("A", 9), ("A", 12),("B", 4), ("B", 10), ("B", 11)])

combineddatardd.combineByKey(lambda v:[v],lambda x,y:x+[y],lambda x,y:x+y).collect()

Выход:

[('A', [3, 9, 12]), ('B', [4, 10, 11])]

1. Определить функцию для объединителя, которая устанавливает аккумулятор в первую пару значений ключа который он встречает внутри раздела, преобразует значение, указанное на этом шаге
2. Определяет функцию, которая слияния нового значения того же ключа с значением аккумулятора, захваченным на шаге 1. Примечание: -конвертировать значение для списка в эта функция как значение аккумулятора была преобразована в список на первом этапе
3. Определить функцию для объединения выходов комбинированных отдельных разделов.

Вы можете использовать метод RDD groupByKey .

Вход:

data = [(1, 'a'), (1, 'b'), (2, 'c'), (2, 'd'), (2, 'e'), (3, 'f')]
rdd = sc.parallelize(data)
result = rdd.groupByKey().collect()

Выход:

[(1, ['a', 'b']), (2, ['c', 'd', 'e']), (3, ['f'])]

Я ударил эту страницу, ища пример Java для той же проблемы. (Если ваш случай похож, вот мой пример)

Трюк - вам нужно группировать ключи.

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

public class SparkMRExample {

    public static void main(String[] args) {
        // spark context initialisation
        SparkConf conf = new SparkConf()
                .setAppName("WordCount")
                .setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);

        //input for testing;
        List<String> input = Arrays.asList("Lorem Ipsum is simply dummy text of the printing and typesetting industry.",
                "Lorem Ipsum has been the industry's standard dummy text ever since the 1500s, when an unknown printer took a galley of type and scrambled it to make a type specimen book.",
                "It has survived not only for centuries, but also the leap into electronic typesetting, remaining essentially unchanged.",
                "It was popularised in the 1960s with the release of Letraset sheets containing Lorem Ipsum passages, and more recently with desktop publishing");
        JavaRDD<String> inputRDD = context.parallelize(input);


        // the map phase of word count example
        JavaPairRDD<String, Integer> mappedRDD =
                inputRDD.flatMapToPair( line ->                      // for this input, each string is a line
                        Arrays.stream(line.split("\\s+"))            // splitting into words, converting into stream
                                .map(word -> new Tuple2<>(word, 1))  // each word is assigned with count 1
                                .collect(Collectors.toList()));      // stream to iterable

        // group the tuples by key
        // (String,Integer) -> (String, Iterable<Integer>)
        JavaPairRDD<String, Iterable<Integer>> groupedRDD = mappedRDD.groupByKey();

        // the reduce phase of word count example
        //(String, Iterable<Integer>) -> (String,Integer)
        JavaRDD<Tuple2<String, Integer>> resultRDD =
                groupedRDD.map(group ->                                      //input is a tuple (String, Iterable<Integer>)
                        new Tuple2<>(group._1,                              // the output key is same as input key
                        StreamSupport.stream(group._2.spliterator(), true)  // converting to stream
                                .reduce(0, (f, s) -> f + s)));              // the sum of counts
        //collecting the RRD so that we can print
        List<Tuple2<String, Integer>> result = resultRDD.collect();
        // print each tuple
        result.forEach(System.out::println);
    }
}

Если вы хотите сделать reduceByKey, где тип в приведенных парах KV отличается от типа исходных пар KV, то можно использовать функцию combineByKey. То, что делает функция, - это взять пары KV и объединить их (по Key) в пары KC, где C - это другой тип, чем V.

В одном задаются 3 функции, createCombiner, mergeValue, mergeCombiners. Первый указывает, как преобразовать тип V в тип C, второй описывает, как объединить тип C с типом V, а последний указывает, как объединить тип C с другим типом C. Мой код создает пары KV:

Определите 3 функции следующим образом:

def Combiner(a):    #Turns value a (a tuple) into a list of a single tuple.
    return [a]

def MergeValue(a, b): #a is the new type [(,), (,), ..., (,)] and b is the old type (,)
    a.extend([b])
    return a

def MergeCombiners(a, b): #a is the new type [(,),...,(,)] and so is b, combine them
    a.extend(b)
    return a

Затем My_KMV = My_KV.combineByKey(Combiner, MergeValue, MergeCombiners)

Лучший ресурс, который я нашел при использовании этой функции: http://abshinn.github.io/python/apache-spark/2014/10/11/using-combinebykey-in-apache-spark/

Как указывали другие, a.append(b) или a.extend(b) return None. Таким образом, reduceByKey(lambda a, b: a.append(b)) возвращает None в первой паре пар KV, а затем не работает во второй паре, потому что None.append (b) терпит неудачу. Вы можете обойти это, указав отдельную функцию:

 def My_Extend(a,b):
      a.extend(b)
      return a

Затем вызовите reduceByKey(lambda a, b: My_Extend(a,b)) (использование лямбда-функции здесь может быть ненужным, но я не проверял этот случай.)