Vanity fair

• 14 years in IT
• 11 years in development
  • Developer
  • Team Lead and even CTO (O_o)
• 1 year in data engineering
• And now building tools for data engineers at JetBrains

Big Data Tools

Story of my love with Kotlin

• Used it in production several weeks before first release
• Was first adopter in Sberbank's production
• Giving talks on it
• Trying to use it EVERYWHERE

My story with data engineering

Data engineering is all about Java, Python and Scala.

And of course I've started with Scala — it's native!

But soon I realized that lots of things could be easier with Kotlin!

Kotlin key benefits

• JVM
• null-aware and null-safe type system
• Extension methods (and values)
• Reified generics
• DSL-building abilities
• Compatible (?) with other JVM ecosystem languages

Note on null-aware type system

Note on null-aware type system

class Z
fun main(){
    val nullZ: Z? = null    val z = Z()    println(nullZ is Z)  // false
    println(z is Z)      // true
    println(nullZ is Z?) // true
    println(z is Z?)     // true
}

Note on null-aware type system

class Z
fun main(){
    val nullZ: Z? = null    val z = Z()    println(nullZ is Z)  // false    println(z is Z)      // true
    println(nullZ is Z?) // true
    println(z is Z?)     // true
}

Note on null-aware type system

class Z
fun main(){
    val nullZ: Z? = null    val z = Z()    println(nullZ is Z)  // false
    println(z is Z)      // true    println(nullZ is Z?) // true    println(z is Z?)     // true}

Extension methods

This already exists in stdlib

Reified generics

• Generics on JVM can't be reified. Ever.
• But there are inline methods in Kotlin. They will be inlined at compile time:

• And if method is inlined we can reify generic at call site!

DSL-building

Extension functions in conjunction with functional arguments allow us to following magic:

And the journey begins

I mean

1. It is alike my surname
2. We need to crossbreed Kotlin and Scala to produce something with best parts of both worlds!
3. And at the start of experiment we don't have any idea on its behavior!

First goal

We want to make standart simple operation on DataFrame to work

First sketch

Fails

Encoder can't (de)serialize

Primitive encoders

Getting real

• Will work for correct Java Beans
• Will work for primitives
• Won't work for data classes

Getting real

Gives us ability to perform first piece of magic:

And it starts to look like DSL!
And we love Spark to be DSL

Type inference!

• Generics are everywhere
• Generics are being erased at runtime
• Need to find some hack

Jackson

public abstract class TypeReference<T> {
    protected final Type _type;    protected TypeReference() {
        Type superClass = getClass().getGenericSuperclass();        if (superClass instanceof Class<?>) {
            // ↑ sanity check, should never happen
            throw new IllegalArgumentException(/* */);
            // ↑ comment that not enough data
        }
        _type = ((ParameterizedType) superClass).getActualTypeArguments()[0];    }

    public Type getType() { return _type; }

Jackson

public abstract class TypeReference<T> {
    protected final Type _type;
    protected TypeReference() {
        Type superClass = getClass().getGenericSuperclass();
        if (superClass instanceof Class<?>) {            // ↑ sanity check, should never happen            throw new IllegalArgumentException();            // ↑ comment that not enough data        }        _type = ((ParameterizedType) superClass).getActualTypeArguments()[0];
    }

    public Type getType() { return _type; }

Now let's translate it to Kotlin

Easy!

And use it…

And it won't work

Because Jackson's hack won't work in Kotlin

By the way, it's the beginning of the story of love to the Monster!

Because it's boring when everything works.

What should I do now? Google!

Sponsor of this horror above:

Alexander Udalov (JetBrains, Kotlin)

This is already deprecated

Because typeOf<T>() introduced

Sidenote about typeOf

• Works in compile time
• Still experimental **
• Much faster then Scala's TypeTagss

** do̳͚n'͓̫̞̝̳t̲̜ ͇u̮̳̲͚̯s̯̬̯e̙͈ ̬i̙̟̹̘̳t̙̻͍ ͇̬̫i̜͖͕n ̪̰̝pr̠͕o̺͉̞͖d͖̺̦͚͔u̟̝̜c͍̺t̬͔̯̝̗̜i̮̣͇͔̻̥o̮̙͉̹͎͍n̰̼: p͕̻̥̪̩̻̱̼o̖͕̖̱͎̬̲̺̝͈̝̟w̜̮̺͉̟̱͎̳̫͙͙̩͙͓̞̖̫ͅe̜̰̘͓̮̯r͖͈̮̦̝͖̱͚ ͔̯̭c͚̱͇̦̯ͅo̰̣͔͚͙͙̜̗r̙̫̰͍̝̤͔̻͙̥̤͎̜̩͕̫͇͖r͉̬̭̬̪̲̙͍͖͕͙̘̣͉u͉̥̟̻̜̖̥̮̜̲̲̻̤̥̭̲̟p̞̲̝̫̫̗̬̺͚͖̤t͍̩̼̮̺̯̬ͅs̺͚̙̙̫̝̠̻̯͙̭

New inference

fun schema(type: KType, map: Map<String, KType> = mapOf()): DataType {
    val primitiveSchema = knownDataTypes[type.classifier]    if (primitiveSchema != null) 
        return KSimpleTypeWrapper(
            primitiveSchema, 
            (type.classifier!! as KClass<*>).java, 
            type.isMarkedNullable
        )
    val klass = type.classifier as? KClass<*> ?: 
        throw IllegalArgumentException("Unsupported type $type")
    val args = type.arguments

New inference

fun schema(type: KType, map: Map<String, KType> = mapOf()): DataType {
    val primitiveSchema = knownDataTypes[type.classifier]
    if (primitiveSchema != null)         return KSimpleTypeWrapper(            primitiveSchema,             (type.classifier!! as KClass<*>).java,             type.isMarkedNullable        )    val klass = type.classifier as? KClass<*> ?: 
        throw IllegalArgumentException("Unsupported type $type")
    val args = type.arguments

New inference

fun schema(type: KType, map: Map<String, KType> = mapOf()): DataType {
    val primitiveSchema = knownDataTypes[type.classifier]
    if (primitiveSchema != null) 
        return KSimpleTypeWrapper(
            primitiveSchema, 
            (type.classifier!! as KClass<*>).java, 
            type.isMarkedNullable
        )
    val klass = type.classifier as? KClass<*> ?: 
        throw IllegalArgumentException("Unsupported type $type")
    val args = type.arguments

New inference

transitiveMerge: remember GenericSignature → RuntimeType

New inference

• KStructField: wrapper over StructField
• KSimpleTypeWrapper: wrapper over primitive type
• KDataTypeWrapper: wrapper over data class

These wrappers hold real classes, nullability etc.

Because Scala can't infer them!

1. Add custom handling methods:

2. Add custom logics when there is predefined schema:

• LOTS of debug here
• Scala is very type safe. Bit codegen is not!
• So use
  • LogLevel.DEBUG
  • "spark.sql.codegen.comments" true to view data flow

Future plans

Support for:

• Scala
  • 2.11
  • 2.13
  • 3 (?)
• Spark
  • 2.3.3
  • 2.4.0+