Apache Beam 官方 Kotlin 示例指南：用 Kotlin 编写更简洁的批流一体数据管道

• 批处理
• 流处理
• 大数据

本指南以 Apache Beam 官方博客《Apache Beam + Kotlin = ❤️》（website/www/site/content/en/blog/beam-kotlin.md）为核心，结合当前仓库中examples/kotlin模块的真实源码与测试，系统讲解：为什么要在 Apache Beam 中使用 Kotlin、官方 Kotlin 示例模块的目录结构与构建方式、Java 与 Kotlin 写法的逐行对比，以及从MinimalWordCount到WindowedWordCount的四个 WordCount 系列管道的完整代码级拆解。读完本文，你将能够读懂并动手运行官方 Kotlin 示例，把熟悉的 Java Beam 管道改写成更简洁、更安全的 Kotlin 版本。

一、为什么 Apache Beam 选择拥抱 Kotlin

Apache Beam 是一个统一的批处理与流处理编程模型，其 Java SDK 是整个生态的基石。而 Kotlin 是一门面向 JVM 的开源、静态类型语言，凭借比 Java 更简洁、更安全的语法，长期被 Android 开发者所青睐。官方博客在 2019 年宣布：Beam SDK 的示例代码正式以 Kotlin 形式随 Java 一起发布，这标志着 Beam 社区开始为 JVM 开发者提供第二种一等公民的表达方式。

需要说明的是：博客发布时仅完成了 WordCount 系列示例，并注明"更多示例正在路上"（At the time of writing this post, only the WordCount samples have been added in Kotlin with more samples underway）。而从当前仓库的实际状态看，该模块已大幅扩充——不仅包含完整的 WordCount 四部曲，还加入了cookbook目录（BigQuery、Join、Combine、Filter 等常用模式）与snippets目录（供 Web 文档引用的代码片段），并有配套的 Kotlin 单元测试。下文所有源码引用均来自当前仓库。

二、官方 Kotlin 示例模块在哪里

Kotlin 示例模块位于仓库的 examples/kotlin 目录，模块描述为 "Apache Beam :: Examples :: Kotlin"。它通过 settings.gradle.kts 中的include(":examples:kotlin")纳入整体构建。

从 examples/kotlin/build.gradle 可以看到其技术构成：

• 使用 Kotlin JVM 插件，版本为1.6.10（org.jetbrains.kotlin.jvm）；
• 直接依赖:sdks:java:core（shadow 配置），因此 Kotlin 示例与 Java 示例共享同一套 Beam SDK 核心 API；
• 依赖:sdks:java:extensions:avro、:sdks:java:io:google-cloud-platform（提供 BigQuery、Pub/Sub IO）、kotlin-stdlib、joda-time、slf4j-api等；
• 运行时依赖:runners:direct-java，即默认使用 DirectRunner 本地执行；
• 为directRunner、flinkRunner、sparkRunner三种 Runner 分别创建了 PreCommit 测试任务，运行**/kotlin/**/*Test.class。

源码目录结构如下（包根org.apache.beam.examples.kotlin）：

examples/kotlin/src/main/java/org/apache/beam/examples/kotlin/ ├── MinimalWordCount.kt # 第一个示例：最简管道 ├── WordCount.kt # 第二个示例：最佳实践（自定义 Options、复合 Transform） ├── DebuggingWordCount.kt # 第三个示例：日志、指标与 PAssert 测试 ├── WindowedWordCount.kt # 第四个示例：窗口、时间戳、流批双模式 ├── common/ # 示例公用工具 │ ├── ExampleOptions.kt │ ├── ExampleBigQueryTableOptions.kt │ ├── ExamplePubsubTopicAndSubscriptionOptions.kt │ ├── ExamplePubsubTopicOptions.kt │ ├── ExampleUtils.kt │ └── WriteOneFilePerWindow.kt ├── cookbook/ # 常用模式：Join / Combine / Filter / Distinct / Trigger └── snippets/ # 供官方文档引用的代码片段

配套测试位于 examples/kotlin/src/test，例如DistinctExampleTest.kt、FilterExamplesTest.kt、JoinExamplesTest.kt、MaxPerKeyExamplesTest.kt。

三、Java 与 Kotlin 写法逐行对比：官方博客的四个经典片段

官方博客通过四组对比片段直观展示 Kotlin 的简洁性。这些片段今天依然可以在仓库源码中找到对应实现，下面完整保留并逐一剖析。

3.1 字符串模板（String Templating）替代String.format

Java 版本需要调用String.format并手工填充占位符：

String filename = String.format( "%s-%s-of-%s%s", filenamePrefixForWindow(intervalWindow), shardNumber, numShards, outputFileHints.suggestedFilenameSuffix);

Kotlin 版本直接用$模板内插，可读性大幅提升：

// String templating val filename = "$filenamePrefixForWindow(intervalWindow)-$shardNumber-of-$numShards${outputFileHints.suggestedFilenameSuffix}"

这段代码并非孤例——在 WriteOneFilePerWindow.kt 的windowedFilename中，实际生产代码正是这样构造每个窗口的分片文件名：

val filename = "${filenamePrefixForWindow(intervalWindow)}-$shardNumber-of-$numShards${outputFileHints.suggestedFilenameSuffix}"

3.2 单行函数体：简化SimpleFunction

Java 中需要显式声明一个继承SimpleFunction<KV<String, Long>, String>的内部类并覆写apply：

public static class FormatAsTextFn extends SimpleFunction<KV<String, Long>, String> { @Override public String apply(KV<String, Long> input) { return input.getKey() + ": " + input.getValue(); } }

Kotlin 利用单表达式函数（single-line functions）把整个类压成一行：

public class FormatAsTextFn : SimpleFunction<KV<String, Long>, String>() { override fun apply(input: KV<String, Long>) = "${input.key} : ${input.value}" //Single line functions }

该类的真实实现见 WordCount.kt，apply返回值会被MapElements.via(FormatAsTextFn())直接消费。

3.3 空安全：?.let消除 null 检查

Java 版本需要显式判空：

if(tableRow != null){ formatAndInsert(tableRow); }

Kotlin 用安全调用与let作用域函数一步完成，且it在块内已被智能转换为非空类型：

tableRow?.let{ formatAndInsert(it) // No need for null checks }

这种空安全哲学贯穿整个 Kotlin 示例模块。例如 ExampleUtils.kt 在检查 BigQuery 表是否存在时，用table?.let { ... } ?: run { ... }优雅地处理"存在则校验 schema、不存在则建表"两种分支；在 ExampleUtils.kt 删除 Pub/Sub topic 时也使用了get(topic)?.let { delete(topic).execute() }。

3.4 类型推断（Type Inferencing）

Java 需要显式写出类型：

String tableName = "testTable";

Kotlin 由编译器自动推断：

val tableName = "testTable" // Type inferencing

类型推断配合val/var让声明更紧凑。在 BigQueryTornadoes.kt 中，arrayListOf<TableFieldSchema>(...)、TableSchema().setFields(fields)等声明均未显式写出泛型类型，却依然保有完整的静态类型检查。

四、第一个示例：MinimalWordCount——最短可运行的管道

MinimalWordCount.kt 是整个系列的起点，无需任何命令行参数，默认以 DirectRunner 本地运行，输出文件名形如wordcounts-00001-of-00005。它用"内联（inline）transform"的方式，把整条管道写成一条链：

public object MinimalWordCount { @JvmStatic fun main(args: Array<String>) { val options = PipelineOptionsFactory.create() val p = Pipeline.create(options) p.apply<PCollection<String>>(TextIO.read().from("gs://apache-beam-samples/shakespeare/*")) .apply( FlatMapElements.into(TypeDescriptors.strings()) .via(ProcessFunction<String, List<String>> { input -> input.split("[^\\p{L}]+".toRegex()).toList() }) ) .apply(Filter.by(SerializableFunction<String, Boolean> { input -> !input.isEmpty() })) .apply(Count.perElement<String>()) .apply( MapElements.into(TypeDescriptors.strings()) .via(ProcessFunction<KV<String, Long>, String> { input -> "${input.key} : ${input.value}" }) ) .apply(TextIO.write().to("wordcounts")) p.run().waitUntilFinish() } }

这段代码集中展示了四个 Beam 核心概念：

1. 读取数据：TextIO.read().from(...)从公开数据集（莎士比亚全集）读取，每个元素是一行文本；
2. 内联 transform：FlatMapElements配合 Kotlin lambda 把每行按[^\p{L}]+（非字母字符）正则切分成单词列表并打平。注意这里使用了 Kotlin 的toRegex()扩展与String.split；
3. 计数：Count.perElement<String>()返回PCollection<KV<String, Long>>，键为单词、值为出现次数；
4. 写出数据：TextIO.write().to("wordcounts")分片写入文本文件。

Filter.by用于过滤空串。由于 Kotlin 函数式接口（SAM）转换的存在，ProcessFunction、SerializableFunction都可以直接用 lambda 实现，代码量比 Java 版本（需要ProcessFunction<String, List<String>>匿名类或单独定义函数）显著更少。

4.1 换个 Runner 跑起来

MinimalWordCount源码注释中给出了切换到其他 Runner 的标准做法（见 MinimalWordCount.kt）：

• 切到 Dataflow：通过options.as(DataflowPipelineOptions::class.java)设置runner = BlockingDataflowRunner，并配置project与tempLocation（GCS 路径）；
• 切到 Flink：options.as(FlinkPipelineOptions::class.java).setRunner(FlinkRunner::class.java)。

五、第二个示例：WordCount——把最佳实践引入管道

WordCount.kt 在最小示例之上引入四个新概念：命令行 Runner 切换、out-of-line 静态 DoFn、复合 Transform、自定义 PipelineOptions。

5.1 自定义 PipelineOptions（Kotlin 接口属性语法）

Beam 的选项系统基于接口属性，而 Kotlin 的var属性声明天然契合这种模式，配合@get:注解目标让代码更干净：

public interface WordCountOptions : PipelineOptions { @get:Description("Path of the file to read from") @get:Default.String("gs://apache-beam-samples/shakespeare/kinglear.txt") var inputFile: String @get:Description("Path of the file to write to") @get:Required var output: String }

对应命令行参数为：

--runner=YOUR_SELECTED_RUNNER # 选择 Runner，如 DirectRunner / FlinkRunner --output=[YOUR_LOCAL_FILE | YOUR_OUTPUT_PREFIX] # 必填（@get:Required 校验） --inputFile=... # 可选，默认读取 King Lear 公开数据集

在main中通过PipelineOptionsFactory.fromArgs(*args).withValidation()解析参数并启用必填项校验。

5.2 静态 DoFn 与复合 Transform

ExtractWordsFn定义在WordCountobject 内部，作为DoFn<String, String>的静态子类，除分词外还使用Metrics.counter与Metrics.distribution统计空行数与行长度分布：

public class ExtractWordsFn : DoFn<String, String>() { private val emptyLines = Metrics.counter(ExtractWordsFn::class.java, "emptyLines") private val lineLenDist = Metrics.distribution(ExtractWordsFn::class.java, "lineLenDistro") @ProcessElement fun processElement(@Element element: String, receiver: DoFn.OutputReceiver<String>) { lineLenDist.update(element.length.toLong()) if (element.trim(' ').isEmpty()) { emptyLines.inc() } val words = element.split(ExampleUtils.TOKENIZER_PATTERN.toRegex()).toTypedArray() for (word in words) { if (word.isNotEmpty()) { receiver.output(word) } } } }

其中分词正则TOKENIZER_PATTERN = "[^\\p{L}]+"定义在 ExampleUtils.kt，\p{L}表示 Unicode 字母类别，即"按一切非字母字符切分"。

CountWords则把"分词 + 计数"打包成可复用的复合 Transform：

public class CountWords : PTransform<PCollection<String>, PCollection<KV<String, Long>>>() { override fun expand(lines: PCollection<String>): PCollection<KV<String, Long>> { val words = lines.apply(ParDo.of(ExtractWordsFn())) val wordCounts = words.apply(Count.perElement()) return wordCounts } }

最终管道组装非常简洁：

p.apply("ReadLines", TextIO.read().from(options.inputFile)) .apply(CountWords()) .apply(MapElements.via(FormatAsTextFn())) .apply<PDone>("WriteCounts", TextIO.write().to(options.output)) p.run().waitUntilFinish()

六、第三个示例：DebuggingWordCount——可观测性与测试

DebuggingWordCount.kt 演示三类生产级实践：分布式日志（SLF4J）、自定义指标（Metrics）、用 PAssert 做管道级断言。

FilterTextFn按正则过滤单词，命中时记录matchedWords计数器并以 DEBUG 级别输出日志，未命中则递增unmatchedWords并以 TRACE 级别记录：

public class FilterTextFn(pattern: String) : DoFn<KV<String, Long>, KV<String, Long>>() { private val filter: Pattern = Pattern.compile(pattern) private val matchedWords = Metrics.counter(FilterTextFn::class.java, "matchedWords") private val unmatchedWords = Metrics.counter(FilterTextFn::class.java, "unmatchedWords") @ProcessElement fun processElement(c: ProcessContext) { if (filter.matcher(c.element().key).matches()) { LOG.debug("Matched: ${c.element().key}") matchedWords.inc() c.output(c.element()) } else { LOG.trace("Did not match: ${c.element().key}") unmatchedWords.inc() } } }

该示例还新增一个选项--filterPattern，默认值为"Flourish|stomach"。管道的点睛之笔是内置断言：

val expectedResults = listOf(KV.of("Flourish", 3L), KV.of("stomach", 1L)) PAssert.that(filteredWords).containsInAnyOrder(expectedResults) p.run().waitUntilFinish()

PAssert是仿照 Hamcrest 集合匹配器设计的测试专用 PTransform：管道成功运行即代表断言通过，无需额外输出。这种"测试即管道"的模式在仓库的 Kotlin 单元测试中被普遍使用。

七、第四个示例：WindowedWordCount——窗口、时间戳与流批双模式

WindowedWordCount.kt 是系列终点，覆盖六个进阶概念：有界/无界输入、时间戳、窗口、窗口化 PCollection 上复用 Transform、访问元素窗口、按窗口分文件写出。

7.1 三个新选项

@get:Description("Fixed window duration, in minutes") @get:Default.Integer(WINDOW_SIZE) // WINDOW_SIZE = 10，默认 10 分钟 var windowSize: Int? @get:Description("Minimum randomly assigned timestamp, in milliseconds-since-epoch") @get:Default.InstanceFactory(DefaultToCurrentSystemTime::class) var minTimestampMillis: Long? @get:Description("Maximum randomly assigned timestamp, in milliseconds-since-epoch") @get:Default.InstanceFactory(DefaultToMinTimestampPlusOneHour::class) var maxTimestampMillis: Long? @get:Description("Fixed number of shards to produce per window") var numShards: Int?

命令行示例：

--output=[YOUR_LOCAL_FILE | YOUR_OUTPUT_PREFIX] --windowSize=15 # 把固定窗口改为 15 分钟 --minTimestampMillis=... # 最小随机时间戳（毫秒） --maxTimestampMillis=... # 最大随机时间戳（毫秒） --numShards=... # 每个窗口的输出分片数

其中DefaultToCurrentSystemTime与DefaultToMinTimestampPlusOneHour都是实现DefaultValueFactory<Long>的类，用于在未传参时动态计算默认值——这展示了 Kotlin 对"选项默认值工厂"模式的直接支持。

7.2 核心管道逻辑

val input = pipeline .apply(TextIO.read().from(options.inputFile)) .apply(ParDo.of(AddTimestampFn(minTimestamp, maxTimestamp))) val windowedWords = input.apply( Window.into<String>(FixedWindows.of(Duration.standardMinutes(options.windowSize!!.toLong())))) val wordCounts = windowedWords.apply(WordCount.CountWords()) // 复用无窗口感知的复合 Transform wordCounts .apply(MapElements.via(WordCount.FormatAsTextFn())) .apply<PDone>(WriteOneFilePerWindow(output, options.numShards))

要点拆解：

• AddTimestampFn为有界输入模拟随机时间戳：在[minTimestamp, maxTimestamp)区间内生成随机Instant，并通过receiver.outputWithTimestamp(element, Instant(randomTimestamp))为元素附加事件时间（见 WindowedWordCount.kt）；
• Window.into(FixedWindows.of(...))按固定窗口分组，默认 10 分钟；
• 由于CountWords只操作KV<String, Long>，对窗口一无所知，因此可以直接复用于窗口化 PCollection——这正是 Beam"窗口透明"设计的体现；
• 结果通过 WriteOneFilePerWindow.kt 写出：其内部PerWindowFiles实现FilenamePolicy，文件名由窗口起止时间-分片号-of-总分片数构成，并启用.withWindowedWrites()，使每个窗口的结果落入独立文件；
• 管道用try { result.waitUntilFinish() } catch (e: Exception) { result.cancel() }保证收到 CTRL-C 等终止信号时取消作业（对应源码注释中的 "The example will try to cancel the pipeline on the signal to terminate the process (CTRL-C)"）。

八、更进一步：cookbook 目录中的常用模式

仓库 README（examples/kotlin/README.md）明确指出：跑完 WordCount 系列后，可以转向cookbook目录学习 Join、Filter、Combine 等实用模式。当前仓库中已包含：

以BigQueryTornadoes.kt为例，它从公开气象数据集读取数据，统计每月龙卷风数量并写回 BigQuery：

interface Options : PipelineOptions { @get:Description("Table to read from, specified as <project_id>:<dataset_id>.<table_id>") @get:Default.String(WEATHER_SAMPLES_TABLE) // apache-beam-testing.samples.weather_stations var input: String @get:Description("Mode to use when reading from BigQuery") @get:Default.Enum("EXPORT") var readMethod: Method @get:Description("BigQuery table to write to, specified as <project_id>:<dataset_id>.<table_id>. The dataset must already exist.") @get:Validation.Required var output: String }

其CountTornadoes复合 Transform 依次完成"提取龙卷风月份 →Count.perElement()计数 → 组装 TableRow"，最终以BigQueryIO.writeTableRows()写出，并通过withCreateDisposition(CREATE_IF_NEEDED)与withWriteDisposition(WRITE_TRUNCATE)控制建表与覆写语义。运行前需先创建输出数据集，命令格式为--output=YOUR_PROJECT_ID:DATASET_ID.TABLE_ID。

九、Kotlin 示例的测试与验证

Kotlin 示例的测试同样用 Kotlin 编写，位于 examples/kotlin/src/test，典型模式是"TestPipeline+Create构造输入 +PAssert断言 +pipeline.run()"。以DistinctExampleTest.kt为例：

@RunWith(JUnit4::class) class DistinctExampleTest { private val pipeline: TestPipeline = TestPipeline.create() @Rule fun pipeline(): TestPipeline = pipeline @Test @Category(ValidatesRunner::class) fun testDistinct() { val strings = listOf("k1", "k5", "k5", "k2", "k1", "k2", "k3") val input = pipeline.apply(Create.of(strings).withCoder(StringUtf8Coder.of())) val output = input.apply(Distinct.create()) PAssert.that(output).containsInAnyOrder("k1", "k5", "k2", "k3") pipeline.run().waitUntilFinish() } @Test @Category(ValidatesRunner::class) fun testDistinctEmpty() { val strings = listOf<String>() val input = pipeline.apply(Create.of(strings).withCoder(StringUtf8Coder.of())) val output = input.apply(Distinct.create()) PAssert.that(output).empty() pipeline.run().waitUntilFinish() } }

此外，examples/kotlin/build.gradle 为每个 PreCommit Runner 生成形如directRunnerPreCommit/flinkRunnerPreCommit/sparkRunnerPreCommit的测试任务，测试类筛选**/kotlin/**/*Test.class，并注入--project、--tempRoot、--runner等 Beam Test Pipeline 选项，用于在 CI 中对 WordCount / WindowedWordCount 等示例做端到端验证。

十、如何构建与运行 Kotlin 示例

在仓库根目录使用 Gradle 即可：

# 编译 Kotlin 示例模块（包含 main 与 test） ./gradlew :examples:kotlin:build # 仅编译 ./gradlew :examples:kotlin:compileKotlin # 运行单元测试（本地 DirectRunner） ./gradlew :examples:kotlin:test # 运行某个 Runner 的 PreCommit 集成测试 ./gradlew :examples:kotlin:directRunnerPreCommit

运行某个示例的main方法时，以WordCount为例（默认读取gs://apache-beam-samples/shakespeare/kinglear.txt，--output为必填项）：

./gradlew :examples:kotlin:run \ -PmainClass=org.apache.beam.examples.kotlin.WordCount \ --args="--runner=DirectRunner --output=/tmp/wordcounts"

需要提示的适用前提：默认输入依赖 GCS 公开数据集，因此运行环境需要能够访问gs://apache-beam-samples；切换 Dataflow / Flink / Spark 等 Runner 时，还需按对应 Runner 的文档配置凭据、临时目录等参数（详见各示例源码注释与 examples/kotlin/README.md）。

十一、总结：从"Java 示例"到"Kotlin 示例"

官方博客的发布宣告了 Beam 在 JVM 语言生态上的多元化。Kotlin 给 Beam 开发者带来的核心收益可以归纳为四点：

1. 更少的样板代码：字符串模板、单表达式函数、类型推断让管道声明与 DoFn 定义大幅瘦身；
2. 更强的空安全：?.let、Elvis 运算符、智能转换让原本需要显式判空的逻辑变得既简洁又受编译器保护；
3. 与 Java SDK 无缝互操作：Kotlin 示例直接复用sdks:java:core的全部 API（Pipeline、ParDo、Count、TextIO、BigQueryIO、PAssert 等），Java 开发者几乎零学习成本迁移；
4. 官方示例与测试双覆盖：examples/kotlin模块不仅提供 WordCount 四部曲与 cookbook 常用模式，还配套 Kotlin 单元测试与多 Runner PreCommit 验证，可直接作为新项目脚手架参考。

对于希望在 Apache Beam 中落地 Kotlin 的团队，最直接的路径就是：阅读本指南提到的源码文件，对照 Java 与 Kotlin 的差异，把自己熟悉的管道逐段改写，并用PAssert编写管道级测试——仓库中所有 Kotlin 示例正是这样一步步从MinimalWordCount演进到WindowedWordCount的。

• 批处理
• 流处理
• 大数据