面向Apache Spark的Kotlin预览版简介

Apache Spark是用于大规模分布式数据处理的开源统一分析引擎。在过去的几年中，已成为最受欢迎的大数据处理工具之一。它涵盖了广泛的任务类型——从数据批处理和简单的ETL(提取/转换/加载)到流和机器学习。

由于Kotlin与Java的互操作性，Kotlin开发者已可以通过Java API使用Apache Spark。但是如此一来，就无法充分发挥Kotlin的潜力，而且通常这让编码变得不够流畅。

今天，我们很高兴地分享Apache Spark的Kotlin API的第一个预览版。这个项目在Kotlin和Apache Spark之间增加了兼容层。允许你使用熟悉的语言特性(例如数据类和lambda表达式)编写符合习惯的Kotlin代码。

Kotlin for Apache Spark还根据部分特性扩展了现有的API。

withSpark和withCached函数

withSpark是使用SparkSession的一种简单而优雅的方法，它将自动为你在代码块末尾调用spark.stop()。 你可以向其传递运行Spark可能需要的参数，例如主位置，日志级别或应用程序名称。它还带有一组方便的默认值，可在本地运行Spark。

这是计算行中字母出现次数的经典示例:

val logFile = "a/path/to/logFile.txt"
withSpark(master = "yarn", logLevel = SparkLogLevel.DEBUG){
    spark.read().textFile(logFile).withCached {
        val numAs = filter { it.contains("a") }.count()
        val numBs = filter { it.contains("b") }.count()
        println("Lines with a: $numAs, lines with b: $numBs")
    }
}

上述例子另一个有用的函数是withCached。在其他API中，如果要将计算分配到多个路径中，但只计算一次，则可以调用’cache’方法。但是会很快变得难以跟踪，你必须记住要取消持久化缓存的数据。否则可能会占用比预期更多的内存，甚至完全破坏内存。 withCached会为你跟踪并持久化。

空安全

Kotlin for Spark在现有方法中添加了leftJoin、rightJoin和其他别名，这是从设计角度的空安全。

fun main() {
​
   data class Coordinate(val lon: Double, val lat: Double)
   data class City(val name: String, val coordinate: Coordinate)
   data class CityPopulation(val city: String, val population: Long)
​
   withSpark(appName = "Find biggest cities to visit") {
       val citiesWithCoordinates = dsOf(
               City("Moscow", Coordinate(37.6155600, 55.7522200)),
           // ...
       )
​
       val populations = dsOf(
               CityPopulation("Moscow", 11_503_501L),
               // ...
       )
       citiesWithCoordinates.rightJoin(populations, citiesWithCoordinates.col("name") <code>==</code> populations.col("city"))
               .filter { (_, citiesPopulation) ->
                   citiesPopulation.population > 15_000_000L
               }
               .map { (city, _) ->
                   // A city may potentially be null in this right join!!!
                   city?.coordinate
               }
               .filterNotNull()
               .show()
   }
}

注意上述示例中的city?.coordinate行。在这个右联接中，城市可能为空。这会在其他JVM Spark API中引起NullPointerException，并且追溯问题的根源非常困难。 Kotlin for Apache Spark会为你处理null安全问题，你可以方便地过滤出null结果。

支持什么

Kotlin for Apache Spark的初始版本支持Apache Spark 3.0，其核心针对Scala 2.12进行了编译。

其API涵盖了创建最适合批处理ETL的自包含Spark应用程序所需的所有方法。

Kotlin for Apache Spark入门

为了帮助你快速开始使用Kotlin for Apache Spark，我们准备了一份快速入门指南，可以帮助你配置环境，正确定义项目的依赖关系并运行第一个以Kotlin编写的自包含Spark应用程序。

下一步

我们知道将现有的框架升级到较新的版本需要一段时间，Spark也不例外。这就是为什么在下一个更新中，我们将添加对早期Spark版本的支持：2.4.2 – 2.4.6。

我们还在Kotlin Spark壳上进行工作，以便你可以以交互方式使用数据并享受探索性数据分析的乐趣。

目前，API并未涵盖Spark Streaming和Spark MLlib，但我们将密切听取你的反馈，并将在我们的路线图中进行相应处理。

将来，我们希望Kotlin能以一等公民的身份加入Apache Spark的正式项目。我们相信，它可以为Kotlin和Spark社区增加价值。这就是为什么我们提出了一个Spark项目改进建议：Kotlin对Apache Spark的支持。我们期待你的意见及讨论。

尝试面向Apache Spark的Kotlin和反馈你的想法！