pyspark - TypeError: объект 'Price' не повторяется - PullRequest
Новая
       6

pyspark - TypeError: объект 'Price' не повторяется

0 голосов
/ 25 августа 2018

Я получаю TypeError, когда пытаюсь преобразовать СДР объекта Price в Pair RDD.

Пример кода: 

priceRDD = pppConformInDF.rdd.map(lambda row: Price(row.vyge_id, row.strm_typ_cd, row.sfb_nm, row.txn_dt, row.vfa_extra_am, '2'))
priceKeyValueRDD = priceRDD.map(lambda price: (",".join([price.vyge_id, price.strm_typ_cd, price.sfb_nm]), list(price)))

Как преобразоватьСДР объекта Price для значения ключа RDD, значением является объект Price, а ключом является комбинация некоторых полей в объекте Price.

Журнал ошибок: 

    priceKeyValueRDD = priceRDD.map(lambda price: (",".join([price.vyge_id, price.strm_typ_cd, price.sfb_nm]), list(price)))
TypeError: 'Price' object is not iterable

    at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:298)
    at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:438)
    at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:421)
    at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:252)
    at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
    at scala.collection.Iterator$class.foreach(Iterator.scala:893)
    at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)
    at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
    at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
    at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
    at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
    at org.apache.spark.InterruptibleIterator.to(InterruptibleIterator.scala:28)
    at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
    at org.apache.spark.InterruptibleIterator.toBuffer(InterruptibleIterator.scala:28)
    at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
    at org.apache.spark.InterruptibleIterator.toArray(InterruptibleIterator.scala:28)
    at org.apache.spark.api.python.PythonRDD$$anonfun$3.apply(PythonRDD.scala:149)
    at org.apache.spark.api.python.PythonRDD$$anonfun$3.apply(PythonRDD.scala:149)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
    at org.apache.spark.scheduler.Task.run(Task.scala:109)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.base/java.lang.Thread.run(Unknown Source)

1 Ответ

0 голосов
/ 25 августа 2018

Вы получили ошибку из-за операции list(price), поскольку цена не может быть преобразована в список.

Просто удалите list, затем используйте groupByKey(), чтобы объединить цены с тем же ключом в список. 

priceRDD = pppConformInDF.rdd.map(lambda row: Price(row.vyge_id, row.strm_typ_cd, row.sfb_nm, row.txn_dt, row.vfa_extra_am, '2'))
priceKeyValueRDD = priceRDD.map(lambda price: (",".join([price.vyge_id, price.strm_typ_cd, price.sfb_nm]), price))
priceGroupedRDD = priceKeyValueRDD.groupByKey() # returns something like [(key->iterable of prices)]
...