Estoy tratando de averiguar si puedo trabajar con Kotlin y Spark, y usar las clases de datos del primero en lugar de las clases de casos de Scala.

Tengo la siguiente clase de datos:

 data class Transaction(var context: String = "", var epoch: Long = -1L, var items: HashSet<String> = HashSet()) : Serializable { companion object { @JvmStatic private val serialVersionUID = 1L } }

Y la parte relevante de la rutina principal se ve así:

 val transactionEncoder = Encoders.bean(Transaction::class.java) val transactions = inputDataset .groupByKey(KeyExtractor(), KeyExtractor.getKeyEncoder()) .mapGroups(TransactionCreator(), transactionEncoder) .collectAsList() transactions.forEach { println("collected Transaction=$it") }

Con TransactionCreator definido como:

 class TransactionCreator : MapGroupsFunction<Tuple2<String, Timestamp>, Row, Transaction> { companion object { @JvmStatic private val serialVersionUID = 1L } override fun call(key: Tuple2<String, Timestamp>, values: MutableIterator<Row>): Transaction { val seq = generateSequence { if (values.hasNext()) values.next().getString(2) else null } val items = seq.toCollection(HashSet()) return Transaction(key._1, key._2.time, items).also { println("inside call Transaction=$it") } } }

Sin embargo, creo que me estoy encontrando con algún tipo de problema de serialización, porque el conjunto termina vacío después de la recolección. Veo el siguiente resultado:

 inside call Transaction=Transaction(context=context1, epoch=1000, items=[c]) inside call Transaction=Transaction(context=context1, epoch=0, items=[a, b]) collected Transaction=Transaction(context=context1, epoch=0, items=[]) collected Transaction=Transaction(context=context1, epoch=1000, items=[])

Probé un KryoRegistrator personalizado para ver si era un problema con el HashSet de Kotlin:

 class MyRegistrator : KryoRegistrator { override fun registerClasses(kryo: Kryo) { kryo.register(HashSet::class.java, JavaSerializer()) // kotlin's HashSet } }

Pero no parece ayudar. ¿Alguna otra idea?

Código completo aquí .