apache-spark - collectAsList 后为空集，即使它在转换运算符中不为空

Question

我想知道我是否可以使用 Kotlin 和 Spark，并使用前者的数据类而不是 Scala 的案例类。

我有以下数据类:

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

主例程的相关部分如下所示:

val transactionEncoder = Encoders.bean(Transaction::class.java)
val transactions = inputDataset
    .groupByKey(KeyExtractor(), KeyExtractor.getKeyEncoder())
    .mapGroups(TransactionCreator(), transactionEncoder)
    .collectAsList()

transactions.forEach { println("collected Transaction=$it") }

TransactionCreator定义为:

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") }
    }
}

但是，我认为我遇到了某种序列化问题，因为集合在收集后最终为空。我看到以下输出:

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=[])

我已经尝试了一个自定义的 KryoRegistrator 来查看它是否是 Kotlin 的 HashSet 的问题:

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

但是好像没什么用。还有其他想法吗？

完整代码 here .

Answer 1

这似乎确实是一个序列化问题。Encoders.bean 的文档指出(Spark v2.4.0):

collection types: only array and java.util.List currently, map support is in progress

将 Transaction 数据类移植到 Java 并将 items 更改为 java.util.List 似乎有所帮助。