Hi i'm using spark streaming in kotlin , such as ...

# datascience

[Icyrockton]

Hi i'm using spark streaming in kotlin , such as mapWithState . but get some Functional interface not serializable exception

java.io.NotSerializableException: streaming.VisitStatisticsKt$$Lambda$1181/1794893473

[Icyrockton]

fun main() {

    withSpark {

        val resetTime = {
            val now = Date()
            val date = Date(now.year, now.month, now.date + 1)
            date.time - now.time
        }
        val javaSparkContext = JavaSparkContext(spark.sparkContext())
        val jssc = JavaStreamingContext(javaSparkContext, Durations.seconds(10))
        jssc.checkpoint("./checkPoint")
        val stream = KafkaUtils.createDirectStream(
            jssc, LocationStrategies.PreferConsistent(),
            ConsumerStrategies.Subscribe<String, String>(
                listOf("test"), kafkaConfig()
            )
        )


        stream.map { Json.decodeFromString<UserLog>(it.value()) }.filter {
            it.action == "visit_station"
        }.mapToPair { Tuple2(it.data["gasStationId"] as Int , 1)  }
            .mapWithState(StateSpec.function(Function3{   /* <----- not serializable interface  */
                    stationId : Int, v2 :Option<Int> , v3 : State<Int> ->
                val accSum = v2.getOrElse { 0 } + v3.option.getOrElse { 0 }
                v3.update(accSum)
                c(stationId , accSum)
            })).print()

        jssc.start()
        jssc.awaitTerminationOrTimeout(resetTime())
    }
}

[altavir]

сс @Pasha Finkelshteyn

[Pasha Finkelshteyn]

@Jolan Rensen [JB] works on it right now!

[Jolan Rensen [JetBrains]]

Yup! I'll test whether it works in the latest version 🙂

[Jolan Rensen [JetBrains]]

More information is in the stacktrace:

Exception in thread "main" java.io.NotSerializableException: DStream checkpointing has been enabled but the DStreams with their functions are not serializable
org.jetbrains.kotlinx.spark.examples.TestKt$main$1$$Lambda$876/0x0000000800753440

[Jolan Rensen [JetBrains]]

I think this is a Spark limitation, not a limitation of the Kotlin Spark Library. I'll see if this works (or not) in Java

[Jolan Rensen [JetBrains]]

or Scala

[Jolan Rensen [JetBrains]]

Same error happens in Java, it's indeed a Spark limitation

[Jolan Rensen [JetBrains]]

should work without checkpointing enabled

[Pasha Finkelshteyn]

Yeah, things should be serializable 🙂 @Icyrockton couls you provide us with working Java/Scala example please?

[Jolan Rensen [JetBrains]]

Okay, it does seem to work if we make an explicitly Serializable lambda, like:

.mapWithState(
    StateSpec.function(
        object : Function3<Int, Option<Int>, State<Int>, Tuple2<Int, Int>>, java.io.Serializable {
            override fun invoke(stationId: Int, v2: Option<Int>, v3: State<Int>): Tuple2<Int, Int> {
                val accSum = v2.getOrElse(0) + v3.getOrElse(0)
                v3.update(accSum)

                return t(stationId, accSum)
            }
        }
    )
)

[Jolan Rensen [JetBrains]]

It's not pretty though, I'll see what I can do to make this better 🙂

[Pasha Finkelshteyn]

It looks like we need Serializable function derivatives…

[Pasha Finkelshteyn]

serializableFunction { /* code of invoke here */ }

[Jolan Rensen [JetBrains]]

yup... Something like that should do the trick

[Jolan Rensen [JetBrains]]

But it's not required everywhere

[Pasha Finkelshteyn]

@Icyrockton looks like you can use workaround described by @Jolan Rensen [JB]

[Jolan Rensen [JetBrains]]

so maybe an alternative StateSpec class could also be an option

[Jolan Rensen [JetBrains]]

@Icyrockton Oh actually, could you try the variant of StateSpec.function with Optional (the Java variant) instead of Option (the Scala variant). That might actually do the trick as well

[Jolan Rensen [JetBrains]]

for me, it compiles like:

.mapWithState<Int, Int, Int, Tuple2<Int, Int>>(
    StateSpec.function { stationId: Int, v2: Optional<Int>, v3: State<Int> ->
        val accSum = v2.getOrElse(0) + v3.getOrElse(0)
        v3.update(accSum)

        t(stationId, accSum)
    }
)

(In theory the first explicit types are not needed, but the scala integration of the kotlin compiler is a bit iffy)

[Jolan Rensen [JetBrains]]

This turns the whole example (for when the next release hits) into:

fun main() {

    val resetTime = {
        val now = Date()
        val date = Date(now.year, now.month, now.date + 1)
        date.time - now.time
    }
    withSparkStreaming(
        batchDuration = Durations.seconds(10),
        timeout = resetTime(),
        checkpointPath = "./checkPoint",
    ) {
        val stream = KafkaUtils.createDirectStream(
            ssc, LocationStrategies.PreferConsistent(),
            ConsumerStrategies.Subscribe<String, String>(
                listOf("test"), kafkaConfig()
            )
        )

        stream
            .map { Json.decodeFromString<UserLog>(it.value()) }
            .filter { it.action == "visit_station" }
            .map { t(it.data["gasStationId"] as Int, 1) }
            .mapWithState(
                StateSpec.function { stationId: Int, v2: Optional<Int>, v3: State<Int> ->
                    val accSum = v2.getOrElse(0) + v3.getOrElse(0)
                    v3.update(accSum)

                    t(stationId, accSum)
                }
            )
            .print()
    }
}

[Icyrockton]

thanks I will try , it's very helpful to me 😄