如何将ojai配置从驱动程序传递给spark中的执行程序？答案

【问题标题】：How to pass ojai configuration from driver to executors in spark?如何将ojai配置从驱动程序传递给spark中的执行程序？
【发布时间】：2021-11-17 15:41:42
【问题描述】：

我想知道如何将 OJAI 连接从 spark 驱动程序传递给它的执行程序。这是我的代码：

val connection = DriverManager.getConnection("ojai:mapr:")
val store = connection.getStore("/tables/table1")
val someStream = messagesDStream.mapPartitions {
  iterator => {
  val list = iterator
    .map(record => record.value())
    .toList
    .asJava
    //TODO serializacja, deserializacja, interface serializable w javie
  val query = connection
    .newQuery()
    .where(connection.newCondition()
      .in("_id", list)
      .build())
    .build()}

我得到的错误：

    Caused by: java.io.NotSerializableException: com.mapr.ojai.store.impl.OjaiConnection
Serialization stack:
        - object not serializable (class: com.mapr.ojai.store.impl.OjaiConnection, value: com.mapr.ojai.store.impl.OjaiConnection@2a367e93)
        - field (class: com.example.App$$anonfun$1, name: connection$1, type: interface org.ojai.store.Connection)
        - object (class com.example.App$$anonfun$1, <function1>)
        at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
        at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
        at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
        at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:342)
        ...

只要与 OJAI 的连接在 mapPartitions 函数内，一切都很好。我知道我需要将配置从驱动程序传递给执行程序才能使代码正常工作，但我不知道该怎么做。楚斯！

【问题讨论】：

here 显示的概念可能有用。

标签： apache-spark apache-spark-sql spark-streaming

【解决方案1】：

您遇到了 spark 最臭名昭著的错误 - 任务不可序列化。本质上，这意味着您尝试序列化的类或对象之一（通过网络从驱动程序发送到执行程序）不能以这种方式处理：这里是 ojai 连接器。

您不能将连接本身从驱动程序传递给执行程序 - 您可以做的，同时避免为来自您的流的每批 RDD 不断重新创建连接，将连接在伴随对象中声明为

@transient 惰性验证连接 = ...

并参考 mapPartitions 中的内容。这将确保每个执行器都与数据库建立连接，该连接将通过多个批次持续存在，因为以这种方式标记的字段不会在驱动程序上创建然后序列化，而是在每个执行器上创建。

【讨论】：