Получение некоторой ошибки при выполнении map / reduce в pyspark RDD

#python #apache-spark #pyspark #rdd

Вопрос:

Я просто пытаюсь изучить PySpark, но меня смущает разница между следующими двумя RDD, я знаю, что один — это набор типов, а другой — список, но оба являются RDD

 rdd = sc.parallelize([('a', 1), ('b', 1), ('a', 3)])
type(rdd)

 rdd = sc.parallelize(['a, 1', 'b, 1', 'a, 3'])
type(rdd)

Код для обработки функций map и reduce:

 priceMap= s.map(lambda o: (o.split(",")[0], float(o.split(",")[1])))
priceMap.reduceByKey(add).take(10)

Я могу легко выполнить функцию map / reduce для вторых данных rdd, но когда я пытаюсь выполнить map или reduce, я получаю следующую ошибку: итак, как мы можем преобразовать первый rdd во вторые данные rdd, или, если есть какой-либо способ устранить следующую ошибку, пожалуйста, помогите. Спасибо

Py4JJavaError: произошла ошибка при вызове z:org.apache.spark.api.python .PythonRDD.runJob.: org.apache.spark.SparkException: задание прервано из-за сбоя этапа: задача 0 на этапе 162.0 не удалась 1 раз, последний сбой: потерянная задача 0.0 на этапе 162.0 (TID 3850, localhost, драйвер исполнителя): org.apache.spark.api.python.Исключение PythonException: трассировка (последний последний вызов):

1. Обратная трассировка предполагает, что существует исключение python. Пожалуйста, покажите свой код map / reduce — ошибка, вероятно, происходит оттуда

2. @mck да, я обновил две строки кода для map и reduce, не могли бы вы мне помочь. Спасибо

3. s и add являются неопределенными переменными

4. поместите имя rdd вместо s, чтобы оно было rdd.map(…. и add — это функция, которую вы можете импортировать, ниже приведена библиотека, пожалуйста, импортируйте ее …… из operator import add

Ответ №1:

Для первого rdd вы можете заменить функцию map:

 rdd = sc.parallelize([('a', 1), ('b', 1), ('a', 3)])
rdd.map(lambda o: (o[0], float(o[1]))).reduceByKey(add).collect()

Это потому split , что работает только со строками, но не с кортежами.