Closed pihanya closed 2 years ago
Good one! will add it along with other streaming functinos next update. Are there any more you found that benefit from having these kinds of extensions?
Jolan, thank you for doing great work in kotlin-spark-api
!
The only reason I have come up with issues is that I saw you revived the process of enhancement of kotlin-spark-api
.
I had to fork kotlin-spark-api
to monkey patch encoders part of kotlin-spark-api
. See details in the spoiler below.
Except for the monkey patch, I think there was nothing special I did in my fork of kotlin-spark-api
.
As the result, I have structurally refactored the library and after refactoring the structure of files appeared to be pretty the same as you did for this repository.
If there will be something to share, be sure that I will create a new issue or make pull requests by myself.
For the monkey patch, it looks like you found a case where there is an encoder in ENCODERS but isSupportedByKotlinClassEncoder
returns true so it isn't used. Would be nice to know for which type this is the case though haha.
Enum support (also in data classes) was added here https://github.com/JetBrains/kotlin-spark-api/pull/99.
I just tested and circular dependencies also don't work in Scala: Exception in thread "main" java.lang.UnsupportedOperationException: cannot have circular references in class, but got the circular reference of class org.jetbrains.kotlinx.spark.examples.DataClass1
Is there any specific use case for where a circular reference is needed?
Would be nice to know for which type this is the case
I guess we will start migrating our product to JetBrains/kotlin-spark-api in a few months. I will let you know if the corner case will be found out. Sorry that I am not able to share it now.
Is there any specific use case for where a circular reference is needed?
Our solution integrates with another app using Kafka and persists all the changes to a data warehouse. From Kafka in a single topic, we get updates of tree structures that resemble folder structures in a file system (see spoiler below) where each update is part of the tree.
For now, we are forced to perform ObjectMapper#readValue
on raw strings inside map { ... }
section what is not quite comfortable and readable in source code.
By the way, the ability to have java.util.UUID
fields in data classes would also be very helpful as all our data comes with UUIDs in it. For now we need to write a lot of boilerplate code that uses UUID#fromString
to instantiate UUID instance from String.
Your specific example doesn't work because List<CatalogueNode>
cannot be encoded. It's an interface which can have functions, values etc. so Spark does not know how to encode that. Only a collection of actual data classes would be allowed. So like
val folderChildren: List<Folder>? = null,
val identityRefChildren: List<IdentityRef>? = null,
But then again, the circular reference appears of course. Unfortunately, I don't think we have a solution for that. Especially since Spark itself does not support circular references. It makes sense if you consider that Datasets are essentially column/row data structures. If circular references were allowed, an infinite recursion could exist within a cell which cannot be saved. Some things I found regarding this: https://issues.apache.org/jira/browse/SPARK-33598
As for java.util.UUID, Spark does not support this as well, so I think it's outside the scope of the Kotlin Spark API to add support for this specific class. Usually we only mirror org.apache.spark.sql.Encoders
It should be mentioned that Spark has its own uuid()
function. Otherwise, it could work to just call UUID#toString
Hi. I'm happy to see this has been added and released.
It does, however, have a name case discrepancy with the built-in function. The original spark function is foreachBatch
while the one in this repository is forEachBatch
. I'm not sure if this is intentional or not, but if it's not intentional, it is perhaps better fixing it sooner rather than later.
@hawkaa, hello! Thank you for the remark.
In Kotlin we follow coding conventions. It is said there: "Names of functions, properties and local variables start with a lowercase letter and use camel case". This is the first argument why forEachBatch
is better option vs foreachBatch
.
Also in kotlin-spark-api we have forEach
function that calls Spark's foreach
under the hood. So forEachBatch
would be more idiomatic (in terms of kotlin-spark-api) comparing to Spark's foreachBatch
.
Therefore, if choosing function naming between forEachBatch
vs foreachBatch
for kotlin-spark-api it is more propriate to use the first variant, that is available in release v1.1.0
.
Plus, as a bonus, there is no function overload issue now :) We do have those for reduce {}
-> reduceK {}
unfortunately.
I found it difficult to call
DataStreamWriter.foreachBatch
because source code won't compile until explicit construction ofVoidFunction2
is added.So I suggest adding such an extension for
DataStreamWriter
: