added encoders 3.2

.github/workflows/generate_docs.yml

@@ -25,5 +25,6 @@ jobs:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           publish_branch: docs
           publish_dir: ./kotlin-spark-api/3.2/target/dokka
+          force_orphan: true
 
 

README.md

@@ -27,14 +27,14 @@ We have opened a Spark Project Improvement Proposal: [Kotlin support for Apache
 - [Code of Conduct](#code-of-conduct)
 - [License](#license)
 
-## Supported versions of Apache Spark #TODO
+## Supported versions of Apache Spark
 
 | Apache Spark | Scala |  Kotlin for Apache Spark        |
 |:------------:|:-----:|:-------------------------------:|
 | 3.0.0+       | 2.12  | kotlin-spark-api-3.0:1.0.2    |
 | 2.4.1+       | 2.12  | kotlin-spark-api-2.4_2.12:1.0.2 |
 | 2.4.1+       | 2.11  | kotlin-spark-api-2.4_2.11:1.0.2 |
-| 3.2.0+       | 2.12  | kotlin-spark-api-2.4_2.12:1.0.3 |
+| 3.2.0+       | 2.12  | kotlin-spark-api-3.2:1.0.3 |
 
 ## Releases
 

examples/pom-3.2_2.12.xml

@@ -24,6 +24,11 @@
       <artifactId>spark-sql_${scala.compat.version}</artifactId>
       <version>${spark3.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-streaming_${scala.compat.version}</artifactId>
+      <version>${spark3.version}</version>
+    </dependency>
   </dependencies>
 
   <build>

kotlin-spark-api/3.2/pom_2.12.xml

@@ -36,6 +36,12 @@
       <version>${spark3.version}</version>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-streaming_${scala.compat.version}</artifactId>
+      <version>${spark3.version}</version>
+      <scope>provided</scope>
+    </dependency>
 
     <!-- Test dependencies -->
     <dependency>

kotlin-spark-api/3.2/src/main/kotlin/org/jetbrains/kotlinx/spark/api/ApiV1.kt

@@ -21,29 +21,33 @@
 
 package org.jetbrains.kotlinx.spark.api
 
-import org.apache.hadoop.shaded.org.apache.commons.math3.exception.util.ArgUtils
 import org.apache.spark.SparkContext
-import org.apache.spark.api.java.JavaSparkContext
+import org.apache.spark.api.java.*
 import org.apache.spark.api.java.function.*
 import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.*
 import org.apache.spark.sql.Encoders.*
 import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.streaming.GroupState
 import org.apache.spark.sql.streaming.GroupStateTimeout
 import org.apache.spark.sql.streaming.OutputMode
 import org.apache.spark.sql.types.*
+import org.apache.spark.unsafe.types.CalendarInterval
 import org.jetbrains.kotlinx.spark.extensions.KSparkExtensions
 import scala.Product
 import scala.Tuple2
+import scala.concurrent.duration.`Duration$`
 import scala.reflect.ClassTag
-import scala.reflect.api.TypeTags.TypeTag
+import scala.reflect.api.StandardDefinitions
 import java.beans.PropertyDescriptor
 import java.math.BigDecimal
 import java.sql.Date
 import java.sql.Timestamp
+import java.time.Duration
 import java.time.Instant
 import java.time.LocalDate
+import java.time.Period
 import java.util.*
 import java.util.concurrent.ConcurrentHashMap
 import kotlin.Any
@@ -95,10 +99,12 @@ val ENCODERS: Map<KClass<*>, Encoder<*>> = mapOf(
     String::class to STRING(),
     BigDecimal::class to DECIMAL(),
     Date::class to DATE(),
-    LocalDate::class to LOCALDATE(), // 3.0 only
+    LocalDate::class to LOCALDATE(), // 3.0+
     Timestamp::class to TIMESTAMP(),
-    Instant::class to INSTANT(), // 3.0 only
-    ByteArray::class to BINARY()
+    Instant::class to INSTANT(), // 3.0+
+    ByteArray::class to BINARY(),
+    Duration::class to DURATION(), // 3.2+
+    Period::class to PERIOD(), // 3.2+
 )
 
 
@@ -154,6 +160,18 @@ inline fun <reified T> SparkSession.dsOf(vararg t: T): Dataset<T> =
 inline fun <reified T> List<T>.toDS(spark: SparkSession): Dataset<T> =
     spark.createDataset(this, encoder<T>())
 
+/**
+ * Utility method to create dataset from RDD
+ */
+inline fun <reified T> RDD<T>.toDS(spark: SparkSession): Dataset<T> =
+    spark.createDataset(this, encoder<T>())
+
+/**
+ * Utility method to create dataset from JavaRDD
+ */
+inline fun <reified T> JavaRDDLike<T, *>.toDS(spark: SparkSession): Dataset<T> =
+    spark.createDataset(this.rdd(), encoder<T>())
+
 /**
  * Main method of API, which gives you seamless integration with Spark:
  * It creates encoder for any given supported type T
@@ -177,12 +195,16 @@ fun <T> generateEncoder(type: KType, cls: KClass<*>): Encoder<T> {
     } as Encoder<T>
 }
 
-private fun isSupportedClass(cls: KClass<*>): Boolean =
-    cls.isData
-        || cls.isSubclassOf(Map::class)
-        || cls.isSubclassOf(Iterable::class)
-        || cls.isSubclassOf(Product::class)
-        || cls.java.isArray
+private fun isSupportedClass(cls: KClass<*>): Boolean = when {
+        cls == ByteArray::class -> false // uses binary encoder
+        cls.isData -> true
+        cls.isSubclassOf(Map::class) -> true
+        cls.isSubclassOf(Iterable::class) -> true
+        cls.isSubclassOf(Product::class) -> true
+        cls.java.isArray -> true
+        else -> false
+    }
+
 
 private fun <T> kotlinClassEncoder(schema: DataType, kClass: KClass<*>): Encoder<T> {
     return ExpressionEncoder(
@@ -1192,7 +1214,7 @@ fun schema(type: KType, map: Map<String, KType> = mapOf()): DataType {
                     DoubleArray::class -> typeOf<Double>()
                     BooleanArray::class -> typeOf<Boolean>()
                     ShortArray::class -> typeOf<Short>()
-                    ByteArray::class -> typeOf<Byte>()
+//                    ByteArray::class -> typeOf<Byte>() handled by BinaryType
                     else -> types.getValue(klass.typeParameters[0].name)
                 }
             } else types.getValue(klass.typeParameters[0].name)
@@ -1290,10 +1312,14 @@ private val knownDataTypes: Map<KClass<out Any>, DataType> = mapOf(
     Float::class to DataTypes.FloatType,
     Double::class to DataTypes.DoubleType,
     String::class to DataTypes.StringType,
-    LocalDate::class to `DateType$`.`MODULE$`,
-    Date::class to `DateType$`.`MODULE$`,
-    Timestamp::class to `TimestampType$`.`MODULE$`,
-    Instant::class to `TimestampType$`.`MODULE$`,
+    LocalDate::class to DataTypes.DateType,
+    Date::class to DataTypes.DateType,
+    Timestamp::class to DataTypes.TimestampType,
+    Instant::class to DataTypes.TimestampType,
+    ByteArray::class to DataTypes.BinaryType,
+    Decimal::class to DecimalType.SYSTEM_DEFAULT(),
+    BigDecimal::class to DecimalType.SYSTEM_DEFAULT(),
+    CalendarInterval::class to DataTypes.CalendarIntervalType,
 )
 
 private fun transitiveMerge(a: Map<String, KType>, b: Map<String, KType>): Map<String, KType> {

kotlin-spark-api/3.2/src/main/kotlin/org/jetbrains/kotlinx/spark/api/SparkHelper.kt

@@ -20,6 +20,11 @@
 package org.jetbrains.kotlinx.spark.api
 
 import org.apache.spark.SparkConf
+import org.apache.spark.api.java.JavaRDD
+import org.apache.spark.api.java.JavaRDDLike
+import org.apache.spark.api.java.JavaSparkContext
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.Dataset
 import org.apache.spark.sql.SparkSession.Builder
 import org.apache.spark.sql.UDFRegistration
 import org.jetbrains.kotlinx.spark.api.SparkLogLevel.ERROR
@@ -78,18 +83,38 @@ inline fun withSpark(builder: Builder, logLevel: SparkLogLevel = ERROR, func: KS
             KSparkSession(this).apply {
                 sparkContext.setLogLevel(logLevel)
                 func()
+                spark.stop()
             }
         }
-        .also { it.stop() }
+}
+
+/**
+ * Wrapper for spark creation which copies params from [sparkConf].
+ *
+ * @param sparkConf Sets a list of config options based on this.
+ * @param logLevel Control our logLevel. This overrides any user-defined log settings.
+ * @param func function which will be executed in context of [KSparkSession] (it means that `this` inside block will point to [KSparkSession])
+ */
+@JvmOverloads
+inline fun withSpark(sparkConf: SparkConf, logLevel: SparkLogLevel = ERROR, func: KSparkSession.() -> Unit) {
+    withSpark(
+        builder = SparkSession.builder().config(sparkConf),
+        logLevel = logLevel,
+        func = func,
+    )
 }
 
 /**
  * This wrapper over [SparkSession] which provides several additional methods to create [org.apache.spark.sql.Dataset]
  */
-@Suppress("EXPERIMENTAL_FEATURE_WARNING", "unused")
-inline class KSparkSession(val spark: SparkSession) {
+class KSparkSession(val spark: SparkSession) {
+
+    val sc: JavaSparkContext by lazy { JavaSparkContext(spark.sparkContext) }
+
     inline fun <reified T> List<T>.toDS() = toDS(spark)
     inline fun <reified T> Array<T>.toDS() = spark.dsOf(*this)
     inline fun <reified T> dsOf(vararg arg: T) = spark.dsOf(*arg)
+    inline fun <reified T> RDD<T>.toDS() = toDS(spark)
+    inline fun <reified T> JavaRDDLike<T, *>.toDS() = toDS(spark)
     val udf: UDFRegistration get() = spark.udf()
 }