Bringing Kotlin Back to Apache Spark™

Kotlin Adapter Example for Apache Spark: A Bachelor's Thesis Walkthrough

Apache Spark has long supported Python, Scala, and Java as first-class citizens, but Kotlin developers had a nice option too, in the form of kotlin-spark-api that let them write idiomatic, type-safe Kotlin code against Spark's Dataset API. The problem is that this option doesn't work with Spark Connect, the gRPC-based client-server architecture that Apache Spark now builds on. Anyone wanting idiomatic Kotlin support on a modern Spark cluster is left without a working path. This post walks through why that's the case, and how we rebuilt Kotlin support to fit this new architecture for our thesis — with a particular focus on the design work, metadata governance ideas, and testing process that shaped the final result. The thesis is two-person’s teamwork: Yerold Sanabria Rios was responsible for literature review, abstract, conclusion and original serialization backend implementation; my responsibility was reflection backend implementation, data contracts and metadata coordination for data catalog, and the testing process, including the fixes that came out of it.

Why the old approach stopped working

Historically, Spark ran as one big process. Your application code, third-party libraries, and Spark’s internal engine all lived in the same Java Virtual Machine. The old Kotlin integration took advantage of this by quietly reaching into Spark’s internal machinery and teaching it how to read Kotlin classes.

Spark Connect changed the rules. The client and the server are now separate processes, often on separate machines, talking to each other over a network protocol. The client no longer has any way to peek into the server’s internals. Any Kotlin support going forward would have to be built entirely on the client side, using only information the client itself could gather about its own types.

The design philosophy: do not reinvent the wheel

The research started initially by contacting maintainers of the kotlin-spark-api, proposing our thesis idea. In return, we got to hear the original maintainers’ full story: the project’s pain points, the approaches that had already been tried, and what they hoped an updated version would look like.

Rather than trying to re-implement Spark’s networking layer from scratch, the approach taken here was to treat the official Spark Connect JVM client as a stable foundation and build a Kotlin layer on top of it using extension functions. This “wrapper-bridge” pattern means the project doesn’t need to chase every change Spark makes to its communication protocol, that work is being absorbed upstream, while still giving Kotlin developers the experience they expect: data classes, null safety, and sealed class hierarchies that feel native rather than bolted on.

This decision mattered for a practical reason too. The earlier kotlin-spark-api struggled because keeping its deep internal hooks working with every new Spark release was so labor-intensive. By staying at arm’s length from Spark’s internals and instead focusing on translating Kotlin’s own type information into the schemas Spark expects, the project becomes something that can realistically be maintained over time.

The result is two interchangeable backends sitting on top of that same client:

• A reflection backend, which inspects Kotlin’s KType metadata at runtime and caches the result.

• A serialization backend, which uses kotlinx.serialization‘s compile-time SerialDescriptor to skip runtime introspection entirely.

Both produce the same kind of artifact — a Spark StructType plus a serializer/deserializer pair — they just get there by different routes, with different tradeoffs.

What the bridge actually looks like

Here’s the shape of the whole system, from a Kotlin List<T> on one end to a registered Unity Catalog table on the other:

A few things worth pointing out in that picture:

• Neither backend talks to Spark directly. Both hand a finished StructType and a row source to the official JVM client, which is the only thing that actually speaks gRPC and Protobuf.

• The BackendRouter dashed arrow isn't something the core library does on its own — it's a demo-level pattern, not a hook inside the adapter itself.

• SerializationCache mirrors ReflectionCache field for field (more on that below), including computing its schema lazily from a same-named SchemaInference.kt file on each side — but only the reflection side’s cached schema feeds UnityCatalogIntegrator, the production DDL path. schemaFor() is a separate accessor, used for drift comparison elsewhere in the library, never for DDL.

• There’s a second, genuinely real path into Unity Catalog: SchemaGovernanceTest.kt, a demo/test file, registers schemas from both backends straight to Unity Catalog’s REST API via UnityCatalogRestClient.createTable() — getSparkSchema() for a BigDecimal type-gap case, schemaFor() for everything else — converting the StructType to UC’s column format with a demo-layer helper, toUcColumns(). This bypasses Spark entirely; it’s a direct HTTP call, not spark.sql().

The same cache, twice

Worth calling out directly: SerializationCache isn’t just architecturally similar to ReflectionCache, it’s the same file with the key type swapped. Side by side:

// reflect/ReflectionCache.kt
internal object ReflectionCache {
    private val schemaCache = ConcurrentHashMap<KType, StructType>()
    private val serializerCache = ConcurrentHashMap<KType, RowSerializer>()
    private val deserializerCache = ConcurrentHashMap<KType, RowDeserializer<*>>()

    fun getSchema(kType: KType): StructType =
        schemaCache.getOrPut(kType) { inferSchemaInternal(kType) }
    // getSerializer / getDeserializer follow the same getOrPut shape
}

// serialization/SerializationCache.kt
internal object SerializationCache {
    private val schemaCache = ConcurrentHashMap<SerialDescriptor, StructType>()
    private val serializerCache = ConcurrentHashMap<SerialDescriptor, SparkSerializer<*>>()
    private val deserializerCache = ConcurrentHashMap<SerialDescriptor, SparkDeserializer<*>>()

    fun getSchema(serializer: KSerializer<*>): StructType =
        schemaCache.getOrPut(serializer.descriptor) { inferSparkSchema(serializer.descriptor) }
    // getSparkSerializer / getSparkDeserializer follow the same getOrPut shape
}

Three ConcurrentHashMaps, the same three artifact families (schema, serializer, deserializer), the same getOrPut call shape. The only real difference is the key — KType on one side, SerialDescriptor on the other. That’s also the likely explanation for the ~2x lookup latency gap the benchmarking chapter found between the two caches at the same O(1) complexity class: it isn’t a different algorithm, it’s SerialDescriptor equality costing more than KType equality on otherwise identical code.

From Kotlin types to Spark schemas, in code

The reflection backend’s entry points live in ReflectAPI.kt — this is the real implementation, lightly trimmed:

// reflect/ReflectAPI.kt
inline fun <reified T : Any> List<T>.toDataFrame(
    spark: SparkSession,
    schema: StructType? = null,
): Dataset<Row> = spark.createDataFrameFromKotlinList(this, typeOf<T>(), schema)

inline fun <reified T : Any> Dataset<Row>.toKotlinList(): List<T> =
    this.toKotlinListFromDataFrame(typeOf<T>())

internal fun SparkSession.createDataFrameViaReflectionInternal(
    data: List<Any>,
    kType: KType,
    schema: StructType? = null,
): Dataset<Row> {
    val resolvedSchema = schema ?: ReflectionCache.getSchema(kType)
    if (data.isEmpty()) return this.createDataFrame(emptyList<Row>(), resolvedSchema)
    val serializer = ReflectionCache.getSerializer(kType, resolvedSchema)
    return this.createDataFrame(LazyRowList(data, serializer), resolvedSchema)
}

internal fun <T : Any> Dataset<Row>.toKotlinClassListInternal(kType: KType): List<T> {
    val collectedRows = this.collectAsList()
    val kClass = kType.jvmErasure

    if (kClass.isSealed) {
        // Each row may be a different subclass — resolved per-row via "_type",
        // so the batch-index optimisation below doesn't apply here.
        val typeColumnIndex = this.schema().fieldIndex("_type")
        return collectedRows.map { row ->
            val typeName = row.get(typeColumnIndex) as String
            val subClass = kClass.allLeafSubclasses().find { it.simpleName == typeName }
                ?: error("Unknown subclass type '$typeName'")
            ReflectionCache.getDeserializer<Any>(subClass.createType()).deserialize(row) as T
        }
    }

    // Non-sealed: all rows share one schema — build the column-index map once per batch.
    val deserializer = ReflectionCache.getDeserializer<T>(kType)
    val columnIndices = deserializer.buildColumnIndexMap(this.schema())
    return collectedRows.map { deserializer.deserialize(it, columnIndices) }
}

The reified keyword is what makes this work without asking the caller for a KClass — the compiler inlines the function body at the call site, so the generic type T survives past the JVM’s normal type erasure. In use, it looks exactly like the rest of idiomatic Kotlin:

data class Person(val name: String, val age: Int?)

val people: Dataset<Row> = listOf(
    Person("Alice", 34),
    Person("Bilal", null)
).toDataFrame(spark)

val roundTripped: List<Person> = people.toKotlinList()

ReflectionCache is what makes repeat calls cheap — the reflective walk only happens once per distinct KType. Sealed types take a different path on the way back in: since each row can be a different subclass, the deserializer is resolved per-row from the _type column instead of once per batch. Everything else shares one schema, so buildColumnIndexMap runs once for the whole batch and gets reused row by row. LazyRowList is what keeps memory flat on the encode side: rows materialize one at a time as Spark's iterator pulls them, instead of being built up front and held for the whole operation.

The AOT path: kotlinx.serialization

The serialization backend’s entry point in SerializationAPI.kt mirrors the reflection one — same optional schema override, same delegation to a non-inline internal function:

// serialization/SerializationAPI.kt
inline fun <reified T> List<T>.toSerializableDataFrame(
    spark: SparkSession,
    schema: StructType? = null,
): Dataset<Row> = spark.createDataFrameFromSerializable(this, serializer<T>(), schema)

fun <T> SparkSession.createDataFrameFromSerializable(
    data: List<T>,
    serializer: KSerializer<T>,
    schema: StructType? = null,
): Dataset<Row> {
    if (data.isEmpty()) return this.emptyDataFrame()
    val resolvedSchema = schema ?: SerializationCache.getSchema(serializer)
    val sparkSerializer = if (schema != null) {
        SparkSerializer(serializer, resolvedSchema)   // schema override: bypass the cache
    } else {
        SerializationCache.getSparkSerializer(serializer)
    }
    return this.createDataFrame(LazySerializableRowList(data, sparkSerializer), resolvedSchema)
}

SchemaInference.kt walks the resulting SerialDescriptor tree and maps each SerialKind to a Spark type — PrimitiveKind to atomic types, StructureKind to nested StructType/ArrayType/MapType. The one case that needs special handling is polymorphism: Spark's schemas are flat, so a PolymorphicKind.SEALED descriptor gets expanded into a union schema, with a synthetic _type discriminator column at position 0:

// serialization/SchemaInference.kt
private fun inferSealedSchema(descriptor: SerialDescriptor): StructType {
    val fields = mutableListOf<StructField>()
    fields.add(DataTypes.createStructField("_type", DataTypes.StringType, false))

    val seen = linkedMapOf<String, DataType>()
    val valueDescriptor = descriptor.getElementDescriptor(1) // the subtypes container
    for (i in 0 until valueDescriptor.elementsCount) {
        val subtype = valueDescriptor.getElementDescriptor(i)
        for (j in 0 until subtype.elementsCount) {
            val name = subtype.getElementName(j)
            if (name !in seen) seen[name] = inferSparkType(subtype.getElementDescriptor(j))
        }
    }
    seen.forEach { (name, type) -> fields.add(DataTypes.createStructField(name, type, true)) }
    return DataTypes.createStructType(fields.toTypedArray())
}

For a simple nested model:

@Serializable data class Address(val city: String, val zip: Int)
@Serializable data class User(val name: String, val address: Address)

val user = User("Alice", Address("Berlin", 10115))
val df = listOf(user).toSerializableDataFrame(spark)

SparkRowEncoder doesn't flatten Address into User's row, but it also isn't literally a recursive SparkRowEncoder — for anything nested one level deep it hands off to a separate SparkStructEncoder, which collects city/zip into its own list and, on endStructure, wraps them in a schema-less GenericRow and pushes that single value back to the parent via a callback:

// serialization/encoders/SparkRowEncoder.kt (root)
override fun beginStructure(descriptor: SerialDescriptor): CompositeEncoder {
    structureDepth++
    return when (descriptor.kind) {
        StructureKind.LIST -> SparkListEncoder({ addValue(it) }, serializersModule)
        StructureKind.MAP  -> SparkMapEncoder({ addValue(it) }, serializersModule)
        StructureKind.CLASS -> when (descriptor.serialName) {
            "kotlinx.datetime.LocalDate", "kotlinx.datetime.Instant" -> this
            else -> if (structureDepth == 1) this
                    else SparkStructEncoder({ addValue(it) }, serializersModule)
        }
        PolymorphicKind.SEALED -> SparkSealedEncoder(
            { row ->
                (row as Row).let { r ->
                    for (i in 0 until r.size()) addValue(r.get(i))
                }
            },
            serializersModule,
            schema,
        )
        else -> this
    }
}

// serialization/encoders/SparkStructEncoder.kt (nested)
internal class SparkStructEncoder(
    private val addToParent: (Any?) -> Unit,
    override val serializersModule: SerializersModule,
) : AbstractEncoder() {
    private val fieldValues = mutableListOf<Any?>()
    override fun encodeString(value: String) { fieldValues.add(value) }
    override fun encodeInt(value: Int) { fieldValues.add(value) }
    // ...one override per primitive type, all just appending to fieldValues

    override fun endStructure(descriptor: SerialDescriptor) {
        addToParent(GenericRow(fieldValues.toTypedArray()))
    }
}

The schema isn’t carried on that inner GenericRow — only the outermost row gets a GenericRowWithSchema. Nested struct shapes are recovered from the parent field’s type during decoding instead. Nesting deeper just chains more SparkStructEncoders through the same addToParent callback, one per level.

Handling Kotlin’s trickier types

Sealed classes. Since Spark has no concept of a union type, every variant’s fields get merged into one wide, mostly-nullable schema, plus the _type discriminator:

@Serializable
sealed class PaymentEvent {
    @Serializable data class CardCharge(val amount: Double, val last4: String) : PaymentEvent()
    @Serializable data class Refund(val amount: Double, val reason: String) : PaymentEvent()
}

val events = listOf<PaymentEvent>(
    PaymentEvent.CardCharge(42.50, "4242"),
    PaymentEvent.Refund(10.00, "duplicate charge")
)
val df = events.toSerializableDataFrame(spark)

That produces a schema of _type STRING NOT NULL, amount DOUBLE, last4 STRING, reason STRING. The CardCharge row comes out as ("CardCharge", 42.5, "4242", null); the Refund row as ("Refund", 10.0, null, "duplicate charge").

Getting there takes two passes, because the subtype’s own field order has no fixed relationship to the union schema’s column order. SparkSealedEncoder first delegates to a SparkSealedSubtypeEncoder that just captures the active subtype’s fields into a Map<String, Any?> by name; only once that’s done does it walk the union schema and place each captured value at its matching column:

// serialization/encoders/SparkSealedEncoder.kt
override fun endStructure(descriptor: SerialDescriptor) {
    val values = arrayOfNulls<Any?>(sealedSchema.fields().size)
    values[0] = typeName
    sealedSchema.fields().forEachIndexed { i, field ->
        if (i > 0 && field.name() in capturedFields) {
            values[i] = capturedFields[field.name()]
        }
    }
    addToParent(GenericRowWithSchema(values, sealedSchema))
}

Fields belonging to the other subtype are simply never written, so they stay at their arrayOfNulls default. On the way back in, the decoder reads _type first and uses it to pick which subclass constructor to invoke before touching anything else.

Value classes. These compile to their wrapped primitive, so the schema doesn’t get an extra nesting level for them — it just unwraps:

@JvmInline value class UserId(val raw: Long)

@Serializable
data class Order(val id: UserId, val total: Double)

Order.id shows up in Spark as a plain LongType column — the reflection backend's type mapper just keeps unwrapping until it hits something that isn't a value class:

// reflect/SchemaInference.kt
internal fun kotlinTypeToSparkType(kType: KType): DataType {
    val classifier = kType.jvmErasure
    return when {
        classifier.isSubclassOf(Enum::class) -> DataTypes.StringType
        classifier.isSealed || classifier.isData ||
            classifier == Pair::class || classifier == Triple::class ->
            ReflectionCache.getSchema(kType)
        classifier.isValue -> {
            val underlying = classifier.primaryConstructor!!.parameters.first().type
            kotlinTypeToSparkType(underlying)   // recurse until it bottoms out at a real type
        }
        else -> kotlinNonStructuralToSparkType(kType)
    }
}

Encoding calls the getter to unwrap UserId down to its Long; decoding does the reverse, forwarding the stored value back into UserId‘s constructor.

The full coverage of the API examples can be found on the notebook here.

Resolving columns by name, not position

One implementation detail worth showing explicitly, because it’s the thing that ended up mattering most in testing: both decoders resolve fields by column name, not by position — and, like the caches, the two backends arrive at this the same way, independently:

// reflect/RowDeserializer.kt
internal fun buildColumnIndexMap(schema: StructType): IntArray =
    IntArray(parameterExtractors.size) { i ->
        try {
            schema.fieldIndex(parameterExtractors[i].paramName)
        } catch (_: IllegalArgumentException) {
            -1   // column absent — fine if the parameter is nullable, checked at decode time
        }
    }

// serialization/decoders/SparkRowDecoder.kt
private fun buildColumnMap(descriptor: SerialDescriptor) {
    if (columnIndexMap != null) return
    val schema = row.schema()
    columnIndexMap = IntArray(descriptor.elementsCount) { i ->
        try {
            schema.fieldIndex(descriptor.getElementName(i))
        } catch (_: IllegalArgumentException) {
            -1
        }
    }
}

Both build a name → index array once — from the deserializer’s parameter names on the reflection side, from the descriptor’s element names on the serialization side — and then every row in the batch just indexes into that array instead of re-scanning the schema. A -1 entry means the column is missing from this particular Row, which is fine for a nullable field and an error for anything else.

When one backend can’t keep up

The serialization backend is faster on decode (no per-field reflective constructor calls), but it can’t describe everything — things like BigDecimal, Set, Pair/Triple, or generic data classes have no built-in kotlinx.serialization representation. The core library doesn’t hide this by silently falling back on your behalf; instead it’s a pattern you wire up yourself, demonstrated in the repo as a BackendRouter:

// demo/resilience/BackendRouter.kt
object BackendRouter {
    inline fun <reified T : Any> encode(
        data: List<T>,
        spark: SparkSession,
        serializer: KSerializer<T>?,
    ): Pair<Dataset<Row>, SchemaDriftReport?> {
        if (serializer == null) {
            // Known type-gap (BigDecimal, Set<T>, java.time.*, ...) — go straight to reflection.
            return data.toDataFrame(spark) to null
        }
        return try {
            data.toSerializableDataFrame(spark) to null
        } catch (e: Exception) {
            val serializationSchema = schemaFor(serializer)
            val reflectionDf = data.toDataFrame(spark)
            val diffs = SchemaDriftReport.compare(serializationSchema, reflectionDf.schema())
            val report = SchemaDriftReport(trigger = SchemaDriftReport.triggerFrom(diffs), /* ... */)
            reflectionDf to report
        }
    }

    inline fun <reified T : Any> decode(
        df: Dataset<Row>,
        serializer: KSerializer<T>,
    ): RouterResult<T> {
        val diffs = SchemaDriftReport.compare(schemaFor(serializer), df.schema())
        if (diffs.isEmpty()) {
            return RouterResult.SerializationSuccess(df.toSerializableKotlinList<T>())
        }
        // Drift detected — report generated before touching any row data.
        val report = SchemaDriftReport(trigger = SchemaDriftReport.triggerFrom(diffs), /* ... */)
        return try {
            RouterResult.ReflectionFallback(df.toKotlinList<T>(), report)
        } catch (e: Exception) {
            RouterResult.BothFailed(report, e)
        }
    }
}

That last detail matters: nothing in spark.kotlin.reflect or spark.kotlin.serialization needs to change to add this behavior — BackendRouter is built entirely out of the public toDataFrame / toSerializableDataFrame functions plus a try/catch. The decode side mirrors this, comparing schemas before touching any row data, so drift is caught up front rather than mid-batch.

You don’t have to decode anything

It’s worth being explicit about one detail in that diagram: toDataFrame() and toSerializableDataFrame() end at a plain Dataset<Row>. Nothing about it is Kotlin-flavored anymore — it’s the same object Scala, Java, or PySpark would produce, so every native Spark operation works on it directly:

val transactions: Dataset<Row> = orders.toDataFrame(spark)

val highValue = transactions
    .filter(col("amount").gt(1000))
    .groupBy("merchant")
    .agg(sum("amount").alias("total"))

highValue.write
    .format("delta")
    .mode("append")
    .save("/mnt/lake/high_value_transactions")

None of that — filter(), groupBy(), agg(), or the final write — passes back through the Kotlin adapter. The decoder, KFunction.callBy, the generated kotlinx.serialization deserializer, none of it runs. toKotlinList() / toSerializableKotlinList() only get invoked if you explicitly ask for typed Kotlin objects back, which is typically the case when a result needs to return to the application tier rather than land in storage. This mirrors how PySpark is normally used: build a DataFrame, hand it to Spark's own execution engine, and let Spark carry it the rest of the way. The decode path exists for when you need it, not as something every row has to pass through on principle. The decode is implemented as .collectAsList, in other words, it has to fit into memory in its current form.

Letting your code talk to your data catalog

One of the more interesting ideas explored in this work goes beyond just “getting Kotlin objects into Spark.” It asks: what if the type information your Kotlin code already enforces could also feed your organization’s data governance tools?

The approach here treats Kotlin’s type definitions as a kind of executable documentation, as a data contract. Since the system already extracts structural information from Kotlin classes to build Spark schemas, that same information can be turned into standard database table definitions and registered with a catalog (in this case, Unity Catalog) over a simple web API. Neither side has to give up its authority — the compiler still enforces correctness in the code, and the catalog still gives non-technical teams visibility into what data looks like — but the two can be kept honest with each other. The project demonstrates how the serialization backend can hit schema drift, or catch a unsupported type. When that happens, it falls back to the reflection backend and produces a structured drift report — one of MISSING_FIELD, SCD_ADDITION, or TYPE_MISMATCH — naming exactly which fields changed and why. From there, it is possible to wire that report into a data catalog so changes surface automatically, being a deliberate choice left to whoever adopts the library, not something the library forces on you. The coupling is intentionally loose: tightening it is a few lines of integration code. Data catalog integration matters in practice: schema metadata only exists in memory for the lifetime of the job. Once the process ends, it’s gone unless something deliberately persisted it elsewhere.

Loose by design, tightened on demand

The coupling to Unity Catalog specifically is loose on purpose, and that’s a property of the actual function signatures, not just a design intention. Three separate knobs, each usable on its own:

Skip DDL extraction. UnityCatalogIntegrator.generateCreateSQL() is itself just a thin wrapper around getSparkSchema() — the same public accessor the reflection backend exposes on its own:

val schema: StructType = getSparkSchema(typeOf<Product>())   // what UnityCatalogIntegrator calls internally

If all you want is the inferred StructType — to diff against an existing table, to feed your own tooling, to log it — there’s no need to go anywhere near UnityCatalogIntegrator at all. Worth being precise here: the serialization backend has its own accessor, schemaFor(), but it isn’t part of this wrapping — UnityCatalogIntegrator.buildCreateStatement() is never called with a schemaFor() result anywhere in the library. That doesn’t mean schemaFor() is unconnected to catalog registration generally, though — SchemaGovernanceTest.kt uses it to register real tables straight to Unity Catalog via UnityCatalogRestClient.createTable(), a separate REST path covered above. Its other real call sites are schema-drift comparison (SchemaDriftReport.compare(...) inside BackendRouter). buildCreateStatement() itself takes a plain StructType and genuinely doesn’t care where it came from, so wiring schemaFor() into it specifically would still be new glue code you’d write yourself — that particular combination isn’t something the library does.

Skip Unity Catalog entirely. Nothing about toDataFrame() or toSerializableDataFrame() requires a catalog of any kind — as the section above showed, df.write.format("delta").save(path) is a complete, valid use of either backend on its own.

Point the same DDL at a different catalog. buildCreateStatement() has no Unity-specific knowledge baked into it — it takes a table name, a StructType, a format string, and an optional location, and emits plain CREATE TABLE ... USING $format:

// unitycatalog/UnityCatalogIntegrator.kt — the real signature
fun buildCreateStatement(
    tableName: String,
    schema: StructType,
    format: String,
    location: String?,
): String

Swap the table-name prefix (and format, if needed) and the exact same call retargets to a different catalog, with the same Kotlin type as the only source of truth.

“Unity” in the class name describes the scope this thesis targeted, not a constraint baked into the DDL itself — what it builds is standard Spark SQL, valid against whatever catalog the active Spark session is configured for.

One more thing, and this one isn’t by design. The three knobs above are intentional looseness. This is a gap — found by checking whether the serialization backend could use UnityCatalogIntegrator just as cleanly as reflection does, and it can’t, quite:

// unitycatalog/UnityCatalogIntegrator.kt — no KSerializer<T> overload exists
inline fun <reified T : Any> generateCreateSQL(tableName: String, ...): String {
    val schema = getSparkSchema(typeOf<T>())   // always reflection, no way to opt into schemaFor()
    return buildCreateStatement(tableName, schema, format, location)
}

Call UnityCatalogIntegrator.registerTable<Product>(spark, ...) on a @Serializable type and it runs fine — Kotlin reflection doesn't care about the annotation — but it silently derives the DDL via reflection, never via the serializer. The two backends order columns differently:

Reflection orders columns via KClass.memberProperties (alphabetical); serialization preserves SerialDescriptor declaration order. So toSerializableDataFrame() on Person(name, age) writes columns as name, age; registerTable<Person>() registers the table as age, name. Same fields, same types, different column order between what the catalog says the table is and what the serialization backend actually writes. The library’s name-based decoders make this mostly harmless for reads, but it’s a real divergence — the missing schemaFor() gap was unintentional, and should be patched later.

Putting it to the test

Good architecture on paper is one thing; making sure it actually behaves correctly under real-world conditions is another. A large part of this project involved building out a testing process that could catch subtle bugs before they became data quality disasters. As a junior, I haven’t been burned by real production incidents yet — so I larped them instead, deliberately working through failure modes rather than waiting to find them the hard way.

One of the most valuable findings came from testing how the system reads data back in and out of Spark into Kotlin objects. The early approach of serialization backend matched up Kotlin properties with Spark columns by position — the first property in your class mapped to the first column, the second to the second, and so on. The decoder expected the order to stay untouched. That works fine as long as nothing ever changes the order of columns.

But real data systems change column order all the time: a new column gets inserted in the middle of a table, a different tool exports columns in a different order, or a data team adds an audit column without telling anyone downstream. Testing against ten scenarios like these revealed that position-based matching doesn’t just fail loudly sometimes — in several cases it fails silently. Two columns of the same type swap places, and the code happily reads an email address into a username field and vice versa, with no error at all. That’s the kind of bug that can sit undetected in a pipeline for a long time.

The fix was to follow reflection backend’s method and match columns by name instead of position, treating column names as the stable identifier they’re meant to be. To prove this mattered, both the old and new approaches were kept side by side and run through the same battery of tests — one designed to confirm the new approach handles reordering gracefully, and one designed to confirm the old approach breaks in exactly the ways predicted. This design decision relies on one assumption: that renaming a column is treated as a breaking change and documented, while reordering isn’t.

Where this leaves things

The end result is a Kotlin integration that doesn’t depend on Spark’s internals at all, gives Kotlin developers back their data classes, enums, value and sealed classes, null safety, and — as a bonus — can keep a data catalog informed about what your code expects. Just as importantly, the testing process surfaced real failure modes like silent data corruption from column reordering that would be easy to miss until they caused a problem.

The project is open sourced and licensed under Apache 2.0. The API is experimental and far from finished. I welcome forks, experimentation, and issues from anyone who wants to contribute.

---

Apache Spark, Spark, Apache, the Apache feather logo, and the Apache Spark project logo are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries. The thesis, its authors, the thesis project, and DataTribe Collective are not affiliated with The Apache Software Foundation.