Pyspark - 推荐引擎 - + 不支持的操作数类型：'int' 和 'str'

【问题标题】：Pyspark - recommendation engine - unsupported operand type(s) for +: 'int' and 'str'Pyspark - 推荐引擎 - + 不支持的操作数类型：'int' 和 'str'
【发布时间】：2016-12-29 09:10:41
【问题描述】：

我正在尝试使用 Pyspark 2.0 和 Python 3.5 编译以下代码

from pyspark import SparkConf, SparkContext

#----- DEFINE FUNCTIONS


def get_counts_and_averages(ID_and_ratings_tuple):
    nratings = len(ID_and_ratings_tuple[1])
    return ID_and_ratings_tuple[0], (nratings, float(sum(x for x in ID_and_ratings_tuple[1]))/nratings)

#-----Create Spark Context

conf = SparkConf().setMaster("local[*]").setAppName("Recommendation Engine")
sc = SparkContext(conf = conf)

# ---- CREATE USER ID AND RATINGS


new_user_ID = 0

# The format of each line is (userID, movieID, rating)
new_user_ratings = [
     (0,260,4), # Star Wars (1977)
     (0,1,3), # Toy Story (1995)
     (0,16,3), # Casino (1995)
     (0,25,4), # Leaving Las Vegas (1995)
     (0,32,4), # Twelve Monkeys (a.k.a. 12 Monkeys) (1995)
     (0,335,1), # Flintstones, The (1994)
     (0,379,1), # Timecop (1994)
     (0,296,3), # Pulp Fiction (1994)
     (0,858,5) , # Godfather, The (1972)
     (0,50,4) # Usual Suspects, The (1995)
    ]
new_user_ratings_RDD = sc.parallelize(new_user_ratings)
print('New user ratings: %s' % new_user_ratings_RDD.take(10))

# ----- Read in data for small ratings data file and put in formation (userID, movieID, rating)

complete_ratings_raw_data = sc.textFile("file:///sparkcourse/ml-latest-small/ratings.csv")
complete_ratings_raw_data_header = complete_ratings_raw_data.take(1)[0]

complete_ratings_data = complete_ratings_raw_data.filter(lambda line: line!=complete_ratings_raw_data_header)\
    .map(lambda line: line.split(",")).map(lambda tokens: (tokens[0],tokens[1],tokens[2])).cache()



movie_ID_with_ratings_RDD = complete_ratings_data.map(lambda x: (x[1], x[2])).groupByKey()
movie_ID_with_avg_ratings_RDD = movie_ID_with_ratings_RDD.map(get_counts_and_averages)
movie_rating_counts_RDD = movie_ID_with_avg_ratings_RDD.map(lambda x: (x[0], x[1][0]))

print(movie_rating_counts_RDD.take(4))

总而言之，它的作用是从movielens读取一堆数据电影评级数据，并尝试应用一些简单的地图功能。

get_counts_and averages 函数存在问题，因为回溯给出以下信息：

org.ap

ache.spark.api.python.PythonException: Traceback (most recent call last):
  File "c:\spark\python\lib\pyspark.zip\pyspark\worker.py", line 172, in main
  File "c:\spark\python\lib\pyspark.zip\pyspark\worker.py", line 167, in process
  File "c:\spark\python\lib\pyspark.zip\pyspark\serializers.py", line 263, in dump_stream
    vs = list(itertools.islice(iterator, batch))
  File "c:\spark\python\lib\pyspark.zip\pyspark\rdd.py", line 1306, in takeUpToNumLeft
  File "c:/sparkcourse/test-recommendation.py", line 8, in get_counts_and_averages
    return ID_and_ratings_tuple[0], (nratings, float(sum(x for x in ID_and_ratings_tuple[1]))/nratings)
TypeError: unsupported operand type(s) for +: 'int' and 'str'

        at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193)
        at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:234)
        at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152)
        at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:63)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
        at org.apache.spark.scheduler.Task.run(Task.scala:85)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
        at java.lang.Thread.run(Unknown Source)
16/08/22 19:55:24 WARN TaskSetManager: Lost task 0.0 in stage 4.0 (TID 7, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "c:\spark\python\lib\pyspark.zip\pyspark\worker.py", line 172, in main
  File "c:\spark\python\lib\pyspark.zip\pyspark\worker.py", line 167, in process
  File "c:\spark\python\lib\pyspark.zip\pyspark\serializers.py", line 263, in dump_stream
    vs = list(itertools.islice(iterator, batch))
  File "c:\spark\python\lib\pyspark.zip\pyspark\rdd.py", line 1306, in takeUpToNumLeft
  File "c:/sparkcourse/test-recommendation.py", line 8, in get_counts_and_averages
    return ID_and_ratings_tuple[0], (nratings, float(sum(x for x in ID_and_ratings_tuple[1]))/nratings)
TypeError: unsupported operand type(s) for +: 'int' and 'str'

        at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193)
        at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:234)
        at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152)
        at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:63)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
        at org.apache.spark.scheduler.Task.run(Task.scala:85)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
        at java.lang.Thread.run(Unknown Source)

16/08/22 19:55:24 ERROR TaskSetManager: Task 0 in stage 4.0 failed 1 times; aborting job
Traceback (most recent call last):
  File "c:/sparkcourse/test-recommendation.py", line 49, in <module>

  File "c:\spark\python\lib\pyspark.zip\pyspark\rdd.py", line 1310, in take
  File "c:\spark\python\lib\pyspark.zip\pyspark\context.py", line 941, in runJob
  File "c:\spark\python\lib\py4j-0.10.1-src.zip\py4j\java_gateway.py", line 933, in __call__
  File "c:\spark\python\lib\py4j-0.10.1-src.zip\py4j\protocol.py", line 312, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.runJob.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 4.0 failed 1 times, most recent failure: Lost task 0.0 in stage 4.0 (TID 7, localhost): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "c:\spark\python\lib\pyspark.zip\pyspark\worker.py", line 172, in main
  File "c:\spark\python\lib\pyspark.zip\pyspark\worker.py", line 167, in process
  File "c:\spark\python\lib\pyspark.zip\pyspark\serializers.py", line 263, in dump_stream
    vs = list(itertools.islice(iterator, batch))
  File "c:\spark\python\lib\pyspark.zip\pyspark\rdd.py", line 1306, in takeUpToNumLeft
  File "c:/sparkcourse/test-recommendation.py", line 8, in get_counts_and_averages
    return ID_and_ratings_tuple[0], (nratings, float(sum(x for x in ID_and_ratings_tuple[1]))/nratings)
TypeError: unsupported operand type(s) for +: 'int' and 'str'

        at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193)
        at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:234)
        at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152)
        at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:63)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
        at org.apache.spark.scheduler.Task.run(Task.scala:85)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
        at java.lang.Thread.run(Unknown Source)

Driver stacktrace:
        at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1450)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1438)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1437)
        at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
        at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
        at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1437)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811)
        at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811)
        at scala.Option.foreach(Option.scala:257)
        at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:811)
        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1659)
        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1618)
        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1607)
        at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
        at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:632)
        at org.apache.spark.SparkContext.runJob(SparkContext.scala:1871)
        at org.apache.spark.SparkContext.runJob(SparkContext.scala:1884)
        at org.apache.spark.SparkContext.runJob(SparkContext.scala:1897)
        at org.apache.spark.api.python.PythonRDD$.runJob(PythonRDD.scala:441)
        at org.apache.spark.api.python.PythonRDD.runJob(PythonRDD.scala)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
        at java.lang.reflect.Method.invoke(Unknown Source)
        at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:237)
        at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
        at py4j.Gateway.invoke(Gateway.java:280)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:128)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.GatewayConnection.run(GatewayConnection.java:211)
        at java.lang.Thread.run(Unknown Source)
Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "c:\spark\python\lib\pyspark.zip\pyspark\worker.py", line 172, in main
  File "c:\spark\python\lib\pyspark.zip\pyspark\worker.py", line 167, in process
  File "c:\spark\python\lib\pyspark.zip\pyspark\serializers.py", line 263, in dump_stream
    vs = list(itertools.islice(iterator, batch))
  File "c:\spark\python\lib\pyspark.zip\pyspark\rdd.py", line 1306, in takeUpToNumLeft
  File "c:/sparkcourse/test-recommendation.py", line 8, in get_counts_and_averages
    return ID_and_ratings_tuple[0], (nratings, float(sum(x for x in ID_and_ratings_tuple[1]))/nratings)
TypeError: unsupported operand type(s) for +: 'int' and 'str'

        at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193)
        at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:234)
        at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152)
        at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:63)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
        at org.apache.spark.scheduler.Task.run(Task.scala:85)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
        ... 1 more

问题似乎是：TypeError: unsupported operand type(s) for +: 'int' and 'str'，但我无法终生弄清楚发生了什么，因为不应该有任何字符串或整数。

有趣的是，如果我打印出命令：print(movie_ID_with_avg_ratings_RDD.take(4) 它可以工作......

非常感谢任何帮助

【问题讨论】：

    标签： python-3.x pyspark


    【解决方案1】：

    这里

    .map(lambda line: line.split(","))

    line 期望是一个字符串，但它是一个 int 元组。你不需要这张地图。

    【讨论】：

    • 不确定是否正确...我仍然遇到同样的错误。当我打印出complete_ratings_data 行时，如果我删除地图，我会得到格式[('1', ',', '1'), ('1', ',', '2'), ...，如果我保留地图，则会得到[('1', '16', '4.0'), ('1', '24', '1.5'), ...
    • 在这种情况下，您通过文件“ratings.csv”加载的元组是 str 元组（您仍然不需要该映射），与“new_user_ratings”中的元组不同。您需要在映射之前或在地图中将 then 转换为 int，如 .map(lambda tokens: (int(tokens[0]),int(tokens[1]),int(tokens[2])))
    猜你喜欢
    • 2022-07-27
    • 2016-04-15
    • 2012-11-29
    • 2012-12-31
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2014-11-15
    相关资源
    最近更新 更多
    热门标签
    Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode