pyspark 如何在Apache Spark中对整数列表进行排序？

lsmepo6l  于 2023-08-02  发布在  Spark
关注(0)|答案(2)|浏览(160)

最近，我开始使用Apache Spark对大量数据进行排序。
在我最初的测试中，我试图在PySpark上并行排序一个整数列表，但显然在使用**sortByKey（）**方法时发生了错误。

  1. >>> rdd = sc.parallelize([5,3,4,7,6,9])
  2. >>> rdd.sortByKey(True)
  4. 2018-09-02 10:39:42 ERROR Executor:91 - Exception in task 2.0 in stage 1.0 (TID 6)
  5. org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  6.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 230, in main
  7.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 225, in process
  8.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\serializers.py", line 372, in dump_stream
  9.     vs = list(itertools.islice(iterator, batch))
  10.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\util.py", line 55, in wrapper
  11.     return f(*args, **kwargs)
  12.   File "C:\spark-2.3.1-bin-hadoop2.7\python\pyspark\rdd.py", line 690, in <lambda>
  13.     samples = self.sample(False, fraction, 1).map(lambda kv: kv[0]).collect()
  14. TypeError: 'int' object has no attribute '__getitem__'
  16.         at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:298)
  17.         at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:438)
  18.         at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:421)
  19.         at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:252)
  20.         at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
  21.         at scala.collection.Iterator$class.foreach(Iterator.scala:893)
  22.         at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)
  23.         at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
  24.         at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
  25.         at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
  26.         at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
  27.         at org.apache.spark.InterruptibleIterator.to(InterruptibleIterator.scala:28)
  28.         at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
  29.         at org.apache.spark.InterruptibleIterator.toBuffer(InterruptibleIterator.scala:28)
  30.         at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
  31.         at org.apache.spark.InterruptibleIterator.toArray(InterruptibleIterator.scala:28)
  32.         at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:939)
  33.         at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:939)
  34.         at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
  35.         at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
  36.         at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
  37.         at org.apache.spark.scheduler.Task.run(Task.scala:109)
  38.         at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
  39.         at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  40.         at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  41.         at java.lang.Thread.run(Thread.java:748)
  42. 2018-09-02 10:39:42 WARN  TaskSetManager:66 - Lost task 2.0 in stage 1.0 (TID 6, localhost, executor driver): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  43.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 230, in main
  44.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 225, in process
  45.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\serializers.py", line 372, in dump_stream
  46.     vs = list(itertools.islice(iterator, batch))
  47.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\util.py", line 55, in wrapper
  48.     return f(*args, **kwargs)
  49.   File "C:\spark-2.3.1-bin-hadoop2.7\python\pyspark\rdd.py", line 690, in <lambda>
  50.     samples = self.sample(False, fraction, 1).map(lambda kv: kv[0]).collect()
  51. TypeError: 'int' object has no attribute '__getitem__'
  53.         at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:298)
  54.         at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:438)
  55.         at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:421)
  56.         at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:252)
  57.         at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
  58.         at scala.collection.Iterator$class.foreach(Iterator.scala:893)
  59.         at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)
  60.         at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
  61.         at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
  62.         at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
  63.         at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
  64.         at org.apache.spark.InterruptibleIterator.to(InterruptibleIterator.scala:28)
  65.         at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
  66.         at org.apache.spark.InterruptibleIterator.toBuffer(InterruptibleIterator.scala:28)
  67.         at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
  68.         at org.apache.spark.InterruptibleIterator.toArray(InterruptibleIterator.scala:28)
  69.         at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:939)
  70.         at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:939)
  71.         at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
  72.         at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
  73.         at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
  74.         at org.apache.spark.scheduler.Task.run(Task.scala:109)
  75.         at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
  76.         at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  77.         at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  78.         at java.lang.Thread.run(Thread.java:748)
  80. 2018-09-02 10:39:42 ERROR TaskSetManager:70 - Task 2 in stage 1.0 failed 1 times; aborting job
  81. Traceback (most recent call last):
  82.   File "<stdin>", line 1, in <module>
  83.   File "C:\spark-2.3.1-bin-hadoop2.7\python\pyspark\rdd.py", line 690, in sortByKey
  84.     samples = self.sample(False, fraction, 1).map(lambda kv: kv[0]).collect()
  85.   File "C:\spark-2.3.1-bin-hadoop2.7\python\pyspark\rdd.py", line 834, in collect
  86.     sock_info = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
  87.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\py4j-0.10.7-src.zip\py4j\java_gateway.py", line 1257, in __call__
  88.   File "C:\spark-2.3.1-bin-hadoop2.7\python\pyspark\sql\utils.py", line 63, in deco
  89.     return f(*a, **kw)
  90.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\py4j-0.10.7-src.zip\py4j\protocol.py", line 328, in get_return_value
  91. py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
  92. : org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 1.0 failed 1 times, most recent failure: Lost task 2.0 in stage 1.0 (TID 6, localhost, executor driver): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  93.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 230, in main
  94.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 225, in process
  95.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\serializers.py", line 372, in dump_stream
  96.     vs = list(itertools.islice(iterator, batch))
  97.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\util.py", line 55, in wrapper
  98.     return f(*args, **kwargs)
  99.   File "C:\spark-2.3.1-bin-hadoop2.7\python\pyspark\rdd.py", line 690, in <lambda>
  100.     samples = self.sample(False, fraction, 1).map(lambda kv: kv[0]).collect()
  101. TypeError: 'int' object has no attribute '__getitem__'
  103.         at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:298)
  104.         at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:438)
  105.         at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:421)
  106.         at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:252)
  107.         at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
  108.         at scala.collection.Iterator$class.foreach(Iterator.scala:893)
  109.         at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)
  110.         at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
  111.         at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
  112.         at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
  113.         at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
  114.         at org.apache.spark.InterruptibleIterator.to(InterruptibleIterator.scala:28)
  115.         at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
  116.         at org.apache.spark.InterruptibleIterator.toBuffer(InterruptibleIterator.scala:28)
  117.         at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
  118.         at org.apache.spark.InterruptibleIterator.toArray(InterruptibleIterator.scala:28)
  119.         at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:939)
  120.         at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:939)
  121.         at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
  122.         at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
  123.         at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
  124.         at org.apache.spark.scheduler.Task.run(Task.scala:109)
  125.         at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
  126.         at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  127.         at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  128.         at java.lang.Thread.run(Thread.java:748)
  130. Driver stacktrace:
  131.         at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1602)
  132.         at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1590)
  133.         at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1589)
  134.         at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
  135.         at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
  136.         at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1589)
  137.         at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:831)
  138.         at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:831)
  139.         at scala.Option.foreach(Option.scala:257)
  140.         at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:831)
  141.         at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1823)
  142.         at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1772)
  143.         at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1761)
  144.         at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
  145.         at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:642)
  146.         at org.apache.spark.SparkContext.runJob(SparkContext.scala:2034)
  147.         at org.apache.spark.SparkContext.runJob(SparkContext.scala:2055)
  148.         at org.apache.spark.SparkContext.runJob(SparkContext.scala:2074)
  149.         at org.apache.spark.SparkContext.runJob(SparkContext.scala:2099)
  150.         at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:939)
  151.         at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  152.         at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
  153.         at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
  154.         at org.apache.spark.rdd.RDD.collect(RDD.scala:938)
  155.         at org.apache.spark.api.python.PythonRDD$.collectAndServe(PythonRDD.scala:162)
  156.         at org.apache.spark.api.python.PythonRDD.collectAndServe(PythonRDD.scala)
  157.         at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  158.         at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  159.         at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  160.         at java.lang.reflect.Method.invoke(Method.java:498)
  161.         at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
  162.         at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
  163.         at py4j.Gateway.invoke(Gateway.java:282)
  164.         at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
  165.         at py4j.commands.CallCommand.execute(CallCommand.java:79)
  166.         at py4j.GatewayConnection.run(GatewayConnection.java:238)
  167.         at java.lang.Thread.run(Thread.java:748)
  168. Caused by: org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  169.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 230, in main
  170.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 225, in process
  171.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\serializers.py", line 372, in dump_stream
  172.     vs = list(itertools.islice(iterator, batch))
  173.   File "C:\spark-2.3.1-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\util.py", line 55, in wrapper
  174.     return f(*args, **kwargs)
  175.   File "C:\spark-2.3.1-bin-hadoop2.7\python\pyspark\rdd.py", line 690, in <lambda>
  176.     samples = self.sample(False, fraction, 1).map(lambda kv: kv[0]).collect()
  177. TypeError: 'int' object has no attribute '__getitem__'
  179.         at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:298)
  180.         at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:438)
  181.         at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:421)
  182.         at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:252)
  183.         at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
  184.         at scala.collection.Iterator$class.foreach(Iterator.scala:893)
  185.         at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)
  186.         at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
  187.         at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
  188.         at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
  189.         at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
  190.         at org.apache.spark.InterruptibleIterator.to(InterruptibleIterator.scala:28)
  191.         at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
  192.         at org.apache.spark.InterruptibleIterator.toBuffer(InterruptibleIterator.scala:28)
  193.         at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
  194.         at org.apache.spark.InterruptibleIterator.toArray(InterruptibleIterator.scala:28)
  195.         at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:939)
  196.         at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12.apply(RDD.scala:939)
  197.         at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
  198.         at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
  199.         at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
  200.         at org.apache.spark.scheduler.Task.run(Task.scala:109)
  201.         at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
  202.         at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  203.         at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  204.         ... 1 more
  206. >>> 2018-09-02 10:39:42 WARN  TaskSetManager:66 - Lost task 0.0 in stage 1.0 (TID 4, localhost, executor driver): TaskKilled (Stage cancelled)
  207. 2018-09-02 10:39:42 WARN  TaskSetManager:66 - Lost task 3.0 in stage 1.0 (TID 7, localhost, executor driver): TaskKilled (Stage cancelled)
  208. 2018-09-02 10:39:43 WARN  TaskSetManager:66 - Lost task 1.0 in stage 1.0 (TID 5, localhost, executor driver): TaskKilled (Stage cancelled)

字符串
如何使用Apache Spark并行排序整数列表？
我正在使用Windows 10、Java SE Development Kit 8和Python 2.7.15

pyspark

2条答案

按热度按时间
r7knjye2

r7knjye21#

您要查找的方法是sortBy，而不是sortByKey：

  1. rdd.sortBy(lambda x: x)

字符串

赞(0）分享  回复(0）举报  2023-08-02
kq0g1dla

kq0g1dla2#

这里不能真正使用sortByKey，因为它是只在PairRDD上实现的方法。唉，您的RDD不是PairRDD，也就是说，不是键/值对的分布式集合。
你可以使用 sortBy（function） 来代替，在这里你必须提供排序条件。对于您的情况，您应该尝试：

  1. rdd.sortBy(lambda x => x)

字符串

赞(0）分享  回复(0）举报  2023-08-02
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com