Обработка больших данных на Kotlin: неожиданный и мощный тандем

Почему Kotlin — это новый «швейцарский нож» для Big Data?

Когда речь заходит об обработке больших данных, в голову обычно приходят Python, Java или Scala. Но есть один язык, который ворвался в этот мир тихо, но уверенно — Kotlin. И если вы думаете, что Kotlin — это только про Android, вы сильно отстали от жизни.

Kotlin для Big Data — это как найти в старом сарае спортивный автомобиль: неожиданно, мощно и чертовски приятно. Он сочетает лаконичность Python, строгость типов Java и функциональные фишки Scala. А главное — он полностью совместим с JVM-экосистемой. Это значит, что вы можете запускать обработку больших данных на Kotlin, используя все те же Hadoop, Spark или Kafka, но писать в 2 раза меньше кода.

В этой статье мы разберем, как использовать Kotlin для реальных задач Big Data: от написания ETL-пайплайнов до работы с Apache Spark и корутинами для асинхронной обработки. Без воды, только код и практика.

Почему Kotlin лучше Java и Scala для Data Engineer?

Давайте честно: Java слишком многословна. Один дата-класс в Java занимает 30 строк с геттерами, сеттерами, equals и hashCode. В Kotlin — одна строка:

data class User(val id: Long, val name: String, val email: String)

Scala, в свою очередь, имеет мощные функциональные возможности, но её синтаксис часто пугает новичков своей сложностью. Kotlin же предлагает «золотую середину» — функциональщину, которая не ломает мозг.

Ключевые преимущества Kotlin для обработки данных:

• Null-safety — вы забудете о NullPointerException в рантайме.
• Extension functions — можно расширять классы Spark или Kafka без наследования.
• Корутины — легковесная асинхронность для потоковой обработки.
• Type inference — код пишется быстрее, читается легче.
• Java Interop — вы подключаете любую Java-библиотеку и используете её как родную.

Для data engineer Kotlin — это возможность писать продакшен-код, который работает быстро, не глючит и легко поддерживается.

Инструментарий: с чем работаем

Прежде чем писать код, давайте определимся, какие библиотеки и фреймворки мы будем использовать для анализа данных на Kotlin:

• Apache Spark для Kotlin — через Java API. Всё работает «из коробки».
• Kotlin DataFrame — библиотека от JetBrains для работы с табличными данными. Аналог Pandas, но на Kotlin.
• Kotlinx.serialization — для работы с JSON/Parquet.
• Kotlin Coroutines — для асинхронной загрузки и обработки потоков.
• Exposed — ORM для работы с БД.

Теперь перейдем к самому интересному — коду.

Пример 1: ETL-пайплайн на Kotlin и Spark

Допустим, у нас есть логи посещений сайта в формате JSON (10 ГБ). Нужно выгрузить их, отфильтровать только платящих пользователей и сохранить в Parquet. Классическая задача обработки данных Kotlin.

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

fun main() {
    val spark = SparkSession.builder()
        .appName("Kotlin Big Data ETL")
        .master("local[*]")
        .getOrCreate()

    // Читаем JSON
    val logsDF = spark.read()
        .option("multiline", "true")
        .json("hdfs://data/logs/*.json")

    // Фильтруем платящих пользователей
    val payingUsers = logsDF
        .filter(col("is_paid") === true)
        .select("user_id", "event_time", "revenue")

    // Агрегируем: общий доход по пользователям
    val revenueByUser = payingUsers
        .groupBy("user_id")
        .agg(sum("revenue").alias("total_revenue"))

    // Сохраняем в Parquet
    revenueByUser.write()
        .mode("overwrite")
        .parquet("hdfs://data/output/revenue_by_user.parquet")

    println("ETL job completed successfully!")
    spark.stop()
}

Обратите внимание: код практически не отличается от Java-версии Spark, но он чище, короче и без лишних скобок. А главное — вы используете все возможности Kotlin: лямбды, безопасные типы, дата-классы.

Пример 2: Работа с Kotlin DataFrame (альтернатива Pandas)

Если вам нужно быстро проанализировать данные без разворачивания Spark-кластера, используйте Kotlin DataFrame. Это библиотека от JetBrains, которая работает in-memory и идеально подходит для небольших датасетов (до нескольких ГБ).

import org.jetbrains.kotlinx.dataframe.api.*
import org.jetbrains.kotlinx.dataframe.io.*

fun main() {
    // Загружаем CSV
    val df = DataFrame.readCSV("data/sales_2024.csv")

    // Выводим схему
    df.printSchema()

    // Фильтруем продажи > 1000
    val highValueSales = df.filter { it["amount"] > 1000 }

    // Группировка по региону
    val regionStats = highValueSales
        .groupBy { it["region"] }
        .aggregate {
            sum { it["amount"] } into "total_sales"
            count() into "transaction_count"
        }

    // Сортируем по убыванию
    val sorted = regionStats.sortByDesc { it["total_sales"] }

    // Выводим топ-5 регионов
    sorted.head(5).print()
}

Этот код выглядит как работа с Pandas, но с типизацией Kotlin. Никаких ошибок с типами столбцов в рантайме — всё проверяется на этапе компиляции. Это колоссальное преимущество для data scientist, который устал от «cannot compare str and int» в Python.

Пример 3: Асинхронная обработка с корутинами

В реальных системах Big Data часто нужно одновременно читать данные из нескольких источников: Kafka, база данных, API. Корутины Kotlin позволяют делать это без блокировки потоков и без тонны callback-ов.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

suspend fun fetchFromKafka(topic: String): ReceiveChannel<String> = coroutineScope {
    produce {
        // Имитация чтения из Kafka
        for (i in 1..100) {
            send("Message $i from $topic")
            delay(10) // симуляция задержки
        }
    }
}

suspend fun fetchFromApi(url: String): String {
    delay(50) // симуляция запроса
    return "Data from $url"
}

fun main() = runBlocking {
    val kafkaChannel = fetchFromKafka("user-events")
    val apiDeferred = async { fetchFromApi("https://api.example.com/users") }

    // Параллельная обработка
    launch {
        for (msg in kafkaChannel) {
            println("Processing Kafka: $msg")
        }
    }