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