spark中的用户定义变量 - org.apache.spark.SparkException：任务不可序列化

【问题标题】：User Defined Variables in spark - org.apache.spark.SparkException: Task not serializablespark中的用户定义变量 - org.apache.spark.SparkException：任务不可序列化
【发布时间】：2016-09-01 17:54:24
【问题描述】：

我想添加用户定义的变量，这些变量可以在 spark 的 filter 和 map 函数中使用，目前我在尝试这样做时遇到了错误。

编辑：我在 zeppelin notebook 上运行此代码，所以我不必创建单独的类。

我的代码有点像：

val some_value = "qwerty"
val some_other_value = "x.y.z"

data.filter(r => r.getString("a.b.c").equals(some_value))
.map(r => (r.getString(some_other_value)))

请注意，这里的“数据”是一个包含 JSON 的 RDD

我收到以下错误：

org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:304)
at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:294)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:122)
at org.apache.spark.SparkContext.clean(SparkContext.scala:2055)
at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:341)
at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:340)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)
at org.apache.spark.rdd.RDD.withScope(RDD.scala:316)
at org.apache.spark.rdd.RDD.filter(RDD.scala:340)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$$$$$fa17825793f04f8d2edd8765c45e2a6c$$$$wC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:202)......
at $print(<console>)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:1065)
at org.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1346)
at org.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:840)
at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:871)
at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:819)
at org.apache.zeppelin.spark.SparkInterpreter.interpretInput(SparkInterpreter.java:747)
at org.apache.zeppelin.spark.SparkInterpreter.interpret(SparkInterpreter.java:711)
at org.apache.zeppelin.spark.SparkInterpreter.interpret(SparkInterpreter.java:704)
at org.apache.zeppelin.interpreter.ClassloaderInterpreter.interpret(ClassloaderInterpreter.java:57)
at org.apache.zeppelin.interpreter.LazyOpenInterpreter.interpret(LazyOpenInterpreter.java:93)
at org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer$InterpretJob.jobRun(RemoteInterpreterServer.java:312)
at org.apache.zeppelin.scheduler.Job.run(Job.java:171)
at org.apache.zeppelin.scheduler.FIFOScheduler$1.run(FIFOScheduler.java:139)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
at java.util.concurrent.FutureTask.run(FutureTask.java:262)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:178)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:292)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:744)
Caused by: java.io.NotSerializableException: org.apache.spark.SparkContext
Serialization stack:
    - object not serializable (class: org.apache.spark.SparkContext, value: org.apache.spark.SparkContext@7f166a30)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: c, type: class org.apache.spark.SparkContext)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC@65b97194)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC@5e9afdf2)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC@1d72f175)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC@7b44290b)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC@2fd0142d)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC@1812605b)
    - field (class: $iwC$$iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC@40020ce9)
    - field (class: $iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC, $iwC$$iwC$$iwC@3db9c1c)
    - field (class: $iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC)
    - object (class $iwC$$iwC, $iwC$$iwC@6968a929)
    - field (class: $iwC, name: $iw, type: class $iwC$$iwC)
    - object (class $iwC, $iwC@1e8d42e6)
    - field (class: $line25.$read, name: $iw, type: class $iwC)
    - object (class $line25.$read, $line25.$read@6b4256a)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: $VAL5449, type: class $line25.$read)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC@7d6cf781)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC@f4d2716)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC@d8c3a34)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC@7bc0d5a5)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC@74ef7040)
    - field (class: $iwC$$iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC@7e55be5d)
    - field (class: $iwC$$iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC, $iwC$$iwC$$iwC@2f59915e)
    - field (class: $iwC$$iwC, name: $iw, type: class $iwC$$iwC$$iwC)
    - object (class $iwC$$iwC, $iwC$$iwC@36faa9d0)
    - field (class: $iwC, name: $iw, type: class $iwC$$iwC)
    - object (class $iwC, $iwC@54f2aef7)
    - field (class: $line385.$read, name: $iw, type: class $iwC)
    - object (class $line385.$read, $line385.$read@3d8590f3)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$$$$$fa17825793f04f8d2edd8765c45e2a6c$$$$wC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: $VAL5516, type: class $line385.$read)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$$$$$fa17825793f04f8d2edd8765c45e2a6c$$$$wC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$$$$$fa17825793f04f8d2edd8765c45e2a6c$$$$wC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC@419dfc19)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$$$$$fa17825793f04f8d2edd8765c45e2a6c$$$$wC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: $outer, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$$$$$fa17825793f04f8d2edd8765c45e2a6c$$$$wC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$$$$$fa17825793f04f8d2edd8765c45e2a6c$$$$wC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$$$$$fa17825793f04f8d2edd8765c45e2a6c$$$$wC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC@6e994c6d)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$$$$$2e9cf4ebd66898e1aa2d2fdd9497ea7$$$$C$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$1, name: $outer, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$$$$$fa17825793f04f8d2edd8765c45e2a6c$$$$wC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$$$$$2e9cf4ebd66898e1aa2d2fdd9497ea7$$$$C$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$1, <function1>)
    at 

    org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
        at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47)
        at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:101)
        at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:301)
        ... 123 more

编辑 2：最后我得到了解决方案，只需将整个代码包装为 scala 函数并传递参数。

val some_value = "qwerty"
val some_other_value = "x.y.z"

def func (value_1 : String, value_2 : String) : RDD[String] = {
val d = data.filter(r => r.getString("a.b.c").equals(value_2))
.map(r => (r.getString(value_2)))
return d
}

val new_data = func(some_value, some_other_value)

另一种方法是在闭包内使用函数来完成操作。

def myFilter(e: String) = (r: org.apache.avro.generic.GenericRecord) => r.getString("a.b.c").equals(e)
def myGenerator(f: String) = (r: org.apache.avro.generic.GenericRecord) => r.getString(f)

val p = data.filter (myFilter(x)).map(myGenerator(y))

【问题讨论】：

  • 请edit 显示完整的课程和堆栈跟踪
  • Spark 通常的问题是它试图序列化函数图中的所有内容。 some_value 或 some_other_value 是否定义在不可序列化的类中？或者扩展/实现一些不可序列化的东西？
  • @Saket - 我将它们定义为独立变量，而不是在任何类中，所以我不确定这是否是问题所在，但我是新手，仍在学习技巧。我按原样运行了上面的代码。有没有办法序列化一个变量？
  • @cricket_007 : 使用包含堆栈跟踪的完整错误消息进行编辑
  • 你是从 Spark shell 运行这个吗？如果将变量内联到 lambda 本身，您应该能够运行查询。

标签： scala apache-spark


【解决方案1】：

org.apache.spark.SparkException: Task not serializable 异常，这意味着您在转换中使用了对不可序列化类的实例的引用。

注意使用外部字段/方法的闭包 对象（这些将引用整个对象）

例如：

NotSerializable notSerializable = new NotSerializable();
JavaRDD<String> rdd = sc.textFile("/tmp/myfile");
rdd.map(s -> notSerializable.doSomething(s)).collect();

Here are some ideas to fix this error:

  1. 可序列化类
  2. 仅在 map 中传递的 lambda 函数内声明实例。
  3. 将 NotSerializable 对象设为静态对象，并在每台机器上创建一次。
  4. 调用rdd.forEachPartition 并在其中创建NotSerializable 对象，如下所示：

rdd.forEachPartition(iter -> { NotSerializable notSerializable = new NotSerializable();

// ...Now process iterator

});

另外，看看Task not serializable: java.io.NotSerializableException when calling function outside closure only on classes not objects

提示：可以使用 jvm 参数查看有关序列化的详细信息

-Dsun.io.serialization.extendedDebugInfo=true 在 SPARK_JAVA_OPTS

  • 序列化调试器：

    SPARK-5307 引入了 SerializationDebugger，Spark 1.3.0 是第一个使用它的版本。它将序列化路径添加到NotSerializableException。当遇到NotSerializableException 时，调试器访问对象图以查找无法序列化的对象的路径，并构造信息以帮助用户找到对象。例如：

    Serialization stack: - object not serializable (class: testing, value: testing@2dfe2f00) - field (class: testing$$anonfun$1, name: $outer, type: class testing) - object (class testing$$anonfun$1, <function1>)

编辑

我正在直接尝试解决这里的问题...

因为我们要在 JVM 上运行 Spark 应用程序，所以我们有一个 要求为了序列化一个对象，你的类应该 显式扩展特殊的Serializable 接口。当你在 Scala 中 正在声明一个案例类，它会自动扩展Serializable 界面。

1.case class 方法：

object Test extends App { case class MyInputForMapAndFilter(somevalue: String, someothervalue: String) val some_value = "qwerty" val some_other_value = "x.y.z" val myInputForMapAndFilter = MyInputForMapAndFilter(some_value,some_other_value) data.filter(r => r.getString("a.b.c").equals(myInputForMapAndFilter.somevalue)) .map(r => (r.getString(myInputForMapAndFilter.someothervalue))) }

scala 中的案例类默认为serializable

2。您可以尝试另一种方法：您可以将 val 声明为 transient 

@transient val some_value = "qwerty"
@transient val some_other_value = "x.y.z"
data.filter(r => r.getString("a.b.c").equals(some_value))
  .map(r => (r.getString(some_other_value)))

请尝试这个并让我知道输出...请主动提出更多问题/问题

【讨论】：

  • 非常感谢详细的帖子和建议。我尝试了您的建议，但仍然收到序列化错误。我完全按照你描述的方式实现了！
  • 你的实现方式和上面的代码sn-p一样吗？
  • 你在使用 kyro 的 java 序列化吗？
  • 是的，我运行代码的方式和上面一样，我是用Zeppelin notebook来运行这些命令的，所以要不要单独使用kyro的java序列化？
  • 好的，你能把所有的变量都包装到一个对象中并作为过滤器传递吗？
【解决方案2】：

您需要做的是使用广播变量，然后您可以在 map 和 reduce 函数中获取该变量。

例如创建广播变量do：

val broadcastString = sc.broadcast("my string")

并且在您的 map 和 reduce 函数中，您可以使用以下方法获取广播字符串：

val myStr = broadcastString.value

更多信息请看这里：http://spark.apache.org/docs/latest/programming-guide.html#broadcast-variables

【讨论】：

  • 我仍然遇到同样的错误：bString_1: org.apache.spark.broadcast.Broadcast[String] = Broadcast(55) bString_2: org.apache.spark.broadcast.Broadcast[String] = Broadcast(56) bString_a: String = WebSessionStartEvent bString_b: String = body.customProperties.Queue.value org.apache.spark.SparkException: Task not serializable at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:304) at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:294) at org.apache.spark.util.ClosureCleaner$.clean....
  • 很抱歉出现混乱的错误信息，我无法格式化。
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 2023-04-02
  • 1970-01-01
  • 2015-05-31
  • 2016-07-27
  • 2015-08-22
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多
热门标签
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