PySpark RuntimeError：在迭代期间设置更改的大小

Question

我正在运行一个pyspark脚本，在下面遇到了一个错误。由于我的代码“如果len(RDD.Take(1))>0:”，它似乎在说“RuntimeError：在迭代期间设置更改的大小”。我不确定这是不是真正的原因，不知道到底出了什么问题。任何帮助都将不胜感激。

谢谢！

17/03/23 21:54:17 INFO DStreamGraph: Updated checkpoint data for time 1490320070000 ms
17/03/23 21:54:17 INFO JobScheduler: Finished job streaming job 1490320072000 ms.0 from job set of time 1490320072000 ms
17/03/23 21:54:17 INFO JobScheduler: Starting job streaming job 1490320072000 ms.1 from job set of time 1490320072000 ms
17/03/23 21:54:17 ERROR JobScheduler: Error running job streaming job 1490320072000 ms.0
org.apache.spark.SparkException: An exception was raised by Python:
Traceback (most recent call last):
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/streaming/util.py",

第65行，在调用r=self.func（t,*rdds)文件“/usr/lib/spark/python/lib/pysspark.zip/pyspark/streaming/dstream.py”中，第159行，在func=lambda t中，第159行，在_compute_glb_max中，如果len(rdd.take(1))>0:File“/usr/lib/spark/python/lib/pysspark.zip/pyspark/rdd)文件”vars=[x._jbroadcast for x in sc._pickled_broadcast_vars]运行时错误：在迭代期间设置更改的大小

  at org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)
  at org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)
  at org.apache.spark.streaming.api.python.PythonDStream$$anonfun$callForeachRDD$1.apply(PythonDStream.scala:179)
  at org.apache.spark.streaming.api.python.PythonDStream$$anonfun$callForeachRDD$1.apply(PythonDStream.scala:179)
  at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ForEachDStream.scala:51)
  at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
  at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
  at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
  at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:50)
  at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
  at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
  at scala.util.Try$.apply(Try.scala:192)
  at org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)
  at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:254)
  at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:254)
  at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:254)
  at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
  at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:253)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  at java.lang.Thread.run(Thread.java:745)
Traceback (most recent call last):
  File "/home/richard/Documents/spark_code/with_kafka/./mongo_kafka_spark_script.py",

行224,在SSC.AwaitTermination（）中；文件“/usr/lib/spark/python/lib/pyspark.zip/pyspark/streaming/context.py”，第206行，在等待终止文件“/usr/lib/spark/py4j-0.10.4-src.zip/py4j/java_gateway.py”中，第1133行，在调用文件“/usr/lib/spark/py4j-0.10.4-src.zip/pyspark/sql/utils.py”中，第63行，在deco文件“：org.apache.spark.sparkException：Python引发了异常：Traceback（最近一次调用）：文件“/usr/lib/spark/python/lib/pyspark.zip/pyspark/streaming/util.py”，第65行，在调用r=self.func（t,*rdds)文件“/usr/lib/spark/python/lib/pyspark.zip/pyspark/streaming/dstream.py”，第159行，在func=lambda t,rdd:old_func(rdd)文件“roadcast_vars,env,includes=_prepare_for_python_rdd（sc,command）文件“/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py”,行2363,in_prepare_for_python_rdd broadcast_vars=[x._jbroadcast for x in sc._pickled_broadcast_vars]运行时错误：在迭代期间设置更改的大小

  at org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)
  at org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)
  at org.apache.spark.streaming.api.python.PythonDStream$$anonfun$callForeachRDD$1.apply(PythonDStream.scala:179)
  at org.apache.spark.streaming.api.python.PythonDStream$$anonfun$callForeachRDD$1.apply(PythonDStream.scala:179)
  at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ForEachDStream.scala:51)
  at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
  at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
  at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
  at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:50)
  at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
  at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
  at scala.util.Try$.apply(Try.scala:192)
  at org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)
  at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:254)
  at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:254)
  at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:254)
  at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
  at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:253)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  at java.lang.Thread.run(Thread.java:745)

Answer 1

在迭代之间创建广播变量似乎不是最佳实践。如果可能，在需要有状态数据时，始终使用updateStateByKey。