Spark & ​​Drools - Как сериализовать KieBase с Kryo

Question

Я работаю в исключении, пытаясь сериализовать KieBase объект в Spark.

Когда я выполняю следующий код:

val kieBase = kieContainer.getKieBase
val broadcastKieBase = spark.sparkContext.broadcast(kieBase)

ConcurrentModificationException Поднялся

Исключение в потоке "main" com.esotericsoftware.kryo.KryoException: java.util.ConcurrentModificationException

Оглядываясь вокруг, я обнаружил, что это известная проблема, но до сих пор нет решения.

Кто-нибудь знает, как сериализовать KieBase пример с Крио?

Answer 1

Если вы выполняете задание Spark с помощью KryoSerializer, задав .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") или по умолчанию установлен KryoSerializer, выполните следующие действия:

Создайте класс DroolsSerializerRegistration, расширив org.apache.spark.serializer.KryoRegistrator. Зарегистрируйте класс, как показано ниже:

      package com.spark.kryo.serializers

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.serializers.JavaSerializer
import org.apache.spark.serializer.KryoRegistrator

class DroolsSerializerRegistration extends KryoRegistrator {
    override def registerClasses(kryo: Kryo) {
        kryo.register(classOf[org.drools.core.impl.KnowledgeBaseImpl], new JavaSerializer)
   }
}

Затем установите конфигурацию ниже в SparkConf .config("spark.kryo.registrator", "com.spark.kryo.serializers.DroolsSerializerRegistration")

Если вы не уверены, что это не работает из-за org.drools.core.impl.KnowledgeBaseImpl затем установите конфигурацию ниже в SparkConf, чтобы получить точный класс, вызывающий проблему. .config("spark.kryo.registrationRequired", "true")

В некоторых случаях он может предоставить несколько классов, а затем зарегистрировать все классы с помощью com.esotericsoftware.kryo.serializers.JavaSerializer. Как только вы выясните все классы, связанные с Drools, удалите spark.kryo.registrationRequired и зарегистрируйте только те классы, которые связаны с библиотекой Drools, а не все, что не удалось.

Answer 2

Я решил эту проблему, я использую spark-submit, а также мой maven jar. это мой конфиг:

.config("spark.serializer", "org.apache.spark.serializer.JavaSerializer")

Answer 3

KieBase не может быть сериализуемым. Вместо этого создайте одноэлементный класс и включите KieBase в исполнителе. У каждого исполнителя будет свой экземпляр.