企业官网建设流程全解析

做实时数仓或者日志平台的同学，迟早都会遇到这样一个需求：Kafka 里进来一批业务数据，要求按数据内容动态写入 Elasticsearch 的不同索引。比如按平台分、按业务类型分、按天分，甚至按租户分。这就是标题里说的“Spark Structured Streaming 基于数据动态写入 ES Index”。

这个需求看起来很直白，但真落地的时候，你会发现网上能搜到的例子 90% 都是写死一个索引名的，什么df.write.format("es").option("es.resource", "fixed_index")之类的。一旦业务要求“这条数据进 A 索引，那条数据进 B 索引”，很多人的第一反应是懵：ES 的 Sink 配置是静态的，怎么让索引名跟着数据走？

这篇笔记把我自己踩过的坑、最终的实现方案、参数调优和生产排查经验都写清楚，适合正在被这个问题卡住的实时计算开发，也适合那些准备从离线数仓转实时、还没被动态索引教育过的朋友。方案不一定是你唯一的选择，但一定是一条能跑通、好维护、出问题能查的路。

1. 需求拆解：为什么“动态写入”是硬需求

1.1 固定索引方案为什么撑不住

先模拟一个典型场景。某跨平台系统的埋点日志进 Kafka，JSON 里带platform、event_type、event_time这些字段，业务方要求日志按平台分索引：click_android、click_ios、click_web。如果写死一个索引，数据全堆在一起，后续的权限隔离、保留周期、删除策略全都没法做。就算你暂时只有一个业务线，等第二个业务线接入，同样一张表混着两套数据，维护成本直接翻倍。

用固定索引还有一个隐性问题：Elasticsearch 索引的 mapping 是第一次写入数据时自动创建的。如果多个业务的数据结构和查询模式完全不一样，硬塞到同一个索引里，要么 mapping 字段爆炸，要么查询性能互相拖累。数据量上去之后，单索引的 shard 规划和生命周期管理也很难做。

所以“按数据动态决定索引名”不是炫技，是真实业务约束下的必然选择。问题的本质是：Spark Structured Streaming 的 Sink 要像一个路由器一样，看着数据内容来决定把这条记录送到哪个索引。

1.2 动态索引的典型业务场景

我遇到的动态索引需求基本可以归成三类：

• 按维度拆分：比如同一个 Kafka topic 里有多方数据，按platform、tenant_id、data_type等字段拆到不同索引。这种最直接，一个字段决定索引后缀。
• 按时间滚动：日志类数据按天建索引，app_log_20250101、app_log_20250102。好处是可以配合索引模板和生命周期策略，冷数据自动归档或删除。
• 维度+时间组合：最完整的形态，比如app_log_android_20250101。既能按业务隔离，又能按时间淘汰，代价是索引数量会膨胀，下文会专门讲怎么治理。

在这三类场景的基础上，索引名本质上就是一个“前缀 + 业务维度字段 + 时间维度字段”拼接出来的字符串。问题就变成了：怎么在流式作业里把这个拼接逻辑做对、做稳。

1.3 可行方案对比

围绕“动态写入 ES”，社区里常见的做法有三种，我直接给对比表：

我最终选的是foreachBatch方案。原因有三个：第一，foreachBatch给了你一个完整的 DataFrame，可以在这个批里做聚合、去重、过滤，再一次性写出；第二，ES connector 的es.resource虽然不支持动态列，但我们可以把动态索引拆成“每个索引一个固定写操作”，在批内循环执行；第三，批内的数据天然有 Spark 的 checkpoint 兜底，即使 ES 写入失败，重试恢复后不会丢批，也不会重复乱序。

2. 核心方案：foreachBatch 分组写入，把动态索引“拍平”

2.1 整体设计思路

动态索引的本质套路就是四个字：先拆后写。在一个微批里，先把数据按目标索引名分组，然后对每个索引名分别执行一次固定索引的 ES 写入。这样既享受了 ES connector 完整的参数能力，又实现了路由逻辑。

拆开之后，一个动态写入的问题就变成若干个静态写入问题，复杂度直接下降一个量级。

sinkDF.writeStream .foreachBatch { (batchDF, batchId) => batchDF .select("es_index") .distinct() .collect() .foreach { row => // 每个索引一次固定写入 } }

你可能会问：为什么不用foreach直接逐条写？因为逐条写会失去批处理的所有优化能力，比如 bulk 批量提交、重试、吞吐调优全部失效，而且 ES 的写入性能主要就是靠批量请求堆起来的。逐条写不仅慢，出问题时还没法做批量重放。

2.2 从 Kafka 到带索引名的 DataFrame

数据流进来之后，第一件事是把 Kafka 的二进制 value 解析成结构化数据，这一步我用from_json配合显式 schema。为什么不直接用spark.sql.streaming.schemaInference？因为流式场景下 schema 一旦变化，checkpoint 恢复很容易报错，显式声明字段类型可以从源头杜绝一批脏数据。

我习惯在 schema 里先保持原始字段，然后只挑需要的字段参与后续计算。这里还有一个生产经验：解析时先不要马上select("data.*")，先看看有没有空值和脏数据，解析失败的行直接过滤掉，避免写入 ES 时动态 mapping 被脏数据污染。

import org.apache.spark.sql.types._ import org.apache.spark.sql.functions._ import org.apache.spark.sql.streaming.Trigger val schema = new StructType() .add("event_id", StringType) .add("event_time", TimestampType) .add("platform", StringType) .add("event_type", StringType) .add("body", StringType) val rawDF = spark.readStream .format("kafka") .option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092") .option("subscribe", "app_log") .option("startingOffsets", "latest") .option("maxOffsetsPerTrigger", "100000") .load() val parsedDF = rawDF .selectExpr("CAST(value AS STRING) AS json_str") .select(from_json(col("json_str"), schema).alias("data")) .select("data.*") .filter(col("event_id").isNotNull && col("event_time").isNotNull)

2.3 分组写入的完整代码

索引名拼接我放在解析之后、写出之前，用withColumn专门生成一列es_index。以平台+日期为例，我的命名规则是app_log_平台_yyyyMMdd：

val sinkDF = parsedDF .withColumn( "es_index", concat( lit("app_log_"), lower(col("platform")), lit("_"), date_format(col("event_time"), "yyyyMMdd") ) ) sinkDF.writeStream .foreachBatch { (batchDF, batchId) => if (!batchDF.isEmpty) { batchDF.cache() batchDF.select("es_index").distinct().collect().foreach { row => val indexName = row.getString(0) batchDF .filter(col("es_index") === indexName) .drop("es_index") .write .format("es") .option("es.resource", indexName) .option("es.nodes", "es-node1:9200,es-node2:9200") .option("es.nodes.wan.only", "true") .option("es.net.http.auth.user", sys.env("ES_USER")) .option("es.net.http.auth.pass", sys.env("ES_PASS")) .option("es.mapping.id", "event_id") .option("es.batch.size.entries", "5000") .option("es.batch.size.bytes", "5mb") .option("es.index.auto.create", "true") .mode("append") .save() } batchDF.unpersist() } } .outputMode("append") .trigger(Trigger.ProcessingTime("60 seconds")) .option("checkpointLocation", "/data/checkpoint/app_log_es") .start() spark.streams.awaitAnyTermination()

这是一段能直接拿去改的代码，但有几个细节必须讲清楚，不然你会踩一个接一个的坑。

2.4 代码拆解：每段在干什么，为什么这么写

为什么生成es_index列而不是解析时直接改字段？

把索引名单独放一列，逻辑上相当于给每条数据打了一个“目标标签”，后面所有分组逻辑都围绕这列做。如果解析时直接拼好索引名，一旦想调整命名规则，就要重新解析一遍原始 JSON。单独拎出来，改一行withColumn就够了。

为什么cache()用在这个位置？

batchDF.select("es_index").distinct().collect()会触发一次 action，后面每个索引要filter一次，又会触发多次 action。如果不 cache，同一个份数据会被反复从 Kafka 的 source 重新计算，每个批次的执行时间直接翻倍。cache 之后，数据在内存里只算一次，索引遍历只是对同一份数据做筛选。这里有句经验：凡是foreachBatch里对 batchDF 做多次写出的，第一行就是batchDF.cache()，写完必须unpersist()，不然 executor 内存会被缓存的数据占着不放，流式任务跑几天必然 OOM。

为什么es.resource用变量而不是固定字符串？

因为 ES connector 不支持在option里通过数据列动态指定索引，所以我在驱动端把索引名从 batchDF 里面筛出来，循环执行。每个循环里es.resource都是确定的，这就是“把动态问题拍平成多个静态问题”的精髓。

为什么 drop 掉es_index列再写？

es_index只是我们用来路由的内部字段，写入 ES 的文档里不应该包含它。不然每条 ES 文档里都会多一个没意义的字段，浪费存储，动态 mapping 也会被它污染。

为什么用es.mapping.id指定event_id？

这是幂等设计。如果上游 Kafka 因为重试或重复提交把同一条数据发了两次，没有指定_id的话，ES 会生成两条文档，实际上就是脏数据。指定业务主键后，相同 ID 的文档会被覆盖更新，后面再写进来的记录不会产生重复。

3. 写入 ES 的关键参数与调优实战

3.1 ES 连接参数，别在认证上卡半天

ES 的连接参数直接影响任务能不能跑起来，尤其是云上的 ES 服务。很多人在本地连没问题，上了云环境就各种连不上，大多数情况是漏了es.nodes.wan.only。

还有个大坑：ES 版本和 connector 版本不匹配时，写入路径会变。老版本写索引要带type，比如indexName/doc；ES 7.x 之后type废弃，直接写索引名。如果你用的 connector 偏老，遇上 ES 8 集群，写请求会直接报错。版本匹配问题排起来非常费劲，建议初始化项目时就固定好版本组合。

3.2 bulk 写入参数，吞吐是这样调出来的

ES 的写入性能跟 bulk 大小呈强相关，Spark connector 底层也是批量提交，所以es.batch.size.entries和es.batch.size.bytes是调优的核心参数。

默认值是 1000 条和 1mb，对于流式场景太小了，每个批次会发起大量 bulk 请求，把集群的连接池打满。我实践中的推荐起点是：

• es.batch.size.entries调到 5000 到 10000，按单条文档大小调整。文档小可以往大调，文档大就调小，目标是单次 bulk 请求体在 5-15mb。
• es.batch.size.bytes调到5mb或10mb。注意这两个参数是“谁先到先触发提交”，不能只调一个。

另外，es.batch.size.entries调大的前提是 ES 集群写入节点能扛住。如果集群的 bulk 队列积压，出现 429 或EsRejectedExecutionException，说明 ES 侧是瓶颈，这时候不是把 batch 调更大，而是要限速。

推荐配合流式端的maxOffsetsPerTrigger控制每个批次的消费量，让写入速度匹配 ES 集群的消费能力。这个参数在新版 Spark 里叫maxOffsetsPerTrigger，老版只对 Kafka source 有效，注意 Spark 版本对应关系。

3.3 数据一致性与幂等

流式任务最怕的不是慢，是“看起来正常但数据多了一条”。ES 写入默认是index操作，缺点是不幂等，重复数据会重复生成文档。解决办法上面已经提过，用es.mapping.id指定业务主键。

如果数据源本身不保证主键唯一，或者你想用更新语义而不是覆盖，还可以配es.write.operation：

• index：默认，直接写文档，不检查 ID。
• upsert：有 ID 就更新，没有就插入，适合做实时宽表更新。
• update：只对已存在的文档做更新，文档不存在时写入失败。

这三个模式各有各的坑。upsert模式下必须同时配es.mapping.id，不然不知道要更新谁；update模式下如果文档不存在会抛异常，要对数据质量有信心才能用。我的建议是：日志数据用index+ 业务主键，宽表更新场景用upsert，不要一上来就上update。

4. 时间分片、时区和索引生命周期

4.1 按天分片的时区坑

动态索引最常用的维度就是时间，按天生成app_log_20250101这种索引。但时间分片有个特别隐蔽的坑：时区。

date_format(col("event_time"), "yyyyMMdd")用的时区是 Spark session 的时区，默认是服务器本地时区。如果你的服务器是 UTC，而业务在北京时间，那么凌晨 8 点之前的数据都会被算到前一天，索引“肉眼看起来”就是偏了一天的。排查这类问题时看数据明明没问题，实际上每天都会有一批数据落错索引，很难发现。

解决办法有两个，一是启动前设置 session 时区：

spark.conf.set("spark.sql.session.timeZone", "Asia/Shanghai")

二是在数据进入计算之前，把时间字段先转换成本地时间：

val sinkDF = parsedDF .withColumn("event_time_local", from_utc_timestamp(col("event_time"), "Asia/Shanghai")) .withColumn( "es_index", concat(lit("app_log_"), date_format(col("event_time_local"), "yyyyMMdd")) )

第二种方法更稳妥，因为 session 时区影响的是全局，如果你同一个任务里还有其他逻辑依赖 UTC 时间，全局改时区反而会引入新问题。我实际生产环境里用的是第二种：业务时间字段统一转北京时间，计算索引名用本地时间字段，其他逻辑保持不动。

4.2 索引模板与 ILM

动态索引数量是会随时间一直增长的，如果裸奔建索引，每个索引的 shard 数、副本数、mapping 都是 ES 自动生成，集群很容易被“小索引”拖垮。所以动态索引必须搭配索引模板。

开发环境验证时自动建索引没问题，生产环境一定要在 ES 侧先备好模板。模板的作用是：新索引首次创建时自动套用预设的配置，不需要应用代码干预。

一个最简模板长这样：

{ "index_patterns": ["app_log_*"], "template": { "settings": { "number_of_shards": 3, "number_of_replicas": 1, "refresh_interval": "10s" }, "mappings": { "properties": { "event_id": { "type": "keyword" }, "event_time": { "type": "date" }, "platform": { "type": "keyword" }, "body": { "type": "text" } } } } }

注意index_patterns要和你业务代码里的索引前缀一致，我这里的app_log_*跟代码里的lit("app_log_")对应。如果前缀不一致，模板就是摆设，所有配置都不会生效。

时间长了之后，你还要处理索引的淘汰。最常见的手段是 ILM（Index Lifecycle Management），设置一个策略，按天滚动的索引到了 30 天后自动删除。这个策略跟模板配合使用，在模板的 settings 里加上：

"lifecycle": { "name": "app_log_policy" }

这样新索引创建时自动绑定策略，不需要人工干预。按天索引加 ILM，是我认为动态索引方案里“最值得做”的进阶配置，因为它直接决定了集群的长期稳定性。

4.3 监控与性能指标

动态写入场景下，我建议至少盯三个指标：

• 每一批的索引数量：如果某个批触发了超过 50 个索引，那要考虑是不是索引基数太大了。大批次 + 大索引数量会导致批处理时间变长，形成积压。
• ES 集群的 bulk 拒绝率：只要出现EsRejectedExecutionException，说明索引速度超过集群写能力。要么调小maxOffsetsPerTrigger，要么给 ES 扩容，不要硬扛。
• Spark Streaming 的 processedRowsPerSecond：这个值突然下降，第一时间怀疑 ES 写入变慢或者 GC 问题。

这些都是常规的排查方向，真正遇到问题的时候，下面这份速查表能帮你省掉大部分试错时间。

5. 生产踩坑实录：10 个高频问题速查表

把我在真实项目里遇到的高频故障整理成了一张速查表，每个问题都是调优半天才找到原因的，直接贴出来：

5.1 重点坑一：动态索引基数爆炸

collect()拿到一批数据里的所有索引名，这个操作有一个前提假设：索引名是低基数的。按天、按平台、按业务类型，一天最多几十个索引，没问题。但如果你动态索引的维度是用户 ID、订单 ID 这种高基数字段，每个批次几千个索引，collect()会把 driver 内存直接打爆，而且每个索引只写很少的数据，ES 集群会生成大量小分片，很快集群就红了。

这类需求本质上不适合“建索引”这种存储模型，更合理的做法是把高基数维度作为文档字段存到同一个索引里，查询时用 ES 的查询条件去过滤。动态索引适合的永远是“有限的维度组合 + 时间滚动”，这个定位要想清楚。

5.2 重点坑二：checkpoint 一旦变更就恢复失败

foreachBatch方案里 checkpoint 记录的是 Kafka offset 和每个批次的处理状态。如果你改了业务逻辑的 schema，比如给StructType加了字段、改了字段类型，重启任务时反序列化旧状态就可能报ClassNotFoundException或 schema 不兼容，任务直接起不来。

这种问题的根因是状态格式变了，旧的 checkpoint 没法兼容新的代码。解决思路有几个：如果改动不复杂，尽量在原 schema 上加 nullable 字段，保持旧字段类型不变；如果改动大，干脆换一个新的 checkpoint 路径，让任务从最新的 Kafka offset 重新开始消费。生产环境里不要忌讳 “丢弃 checkpoint”，关键是丢之前确认数据对账方式，避免丢数据。

5.3 重点坑三：多个索引串行写的长尾效应

一个批里有索引 A、B、C，循环写出的时候是串行的，A 写完才写 B。如果某个索引恰好数据量很大，批处理时间就被它拉长了，比如 50 个索引里 49 个只要 2 秒，剩下一个要 30 秒，整批就变成 32 秒。

我实践里的处理方式是：先保证串行跑通，再针对长尾优化。优化方向有两个，一是把数据量大的索引单独开一个流处理，二是用 Scala 的并发编程包把多个索引的写出放到固定线程池里并行提交，同时要限制并发度为 ES 节点数的一半左右，防止把 ES 写入队列打爆。并行化的收益明显，但排查问题的复杂度也会上去，建议先串行稳定再优化。

6. 最后分享几点个人体会

做完这个项目之后，我对“动态写入”有了一个更深的体会：它在技术上并不复杂，难点全在对“数据路由”的理解和边界条件的把控。索引名拼接、时区转换、幂等键、索引模板，这些单拎出来都不是难题，但组合在一起，每一个都能成为生产事故的导火索。

我个人最满意的设计是“索引模板 + ILM + 业务主键”这套铁三角：模板控制新索引的形态，ILM 控制索引的寿命，主键保证数据的最终一致。这三件事做好，动态写入才算真正具备上线条件。

如果你要复刻这个方案，我的建议是先把每个模块拆开验证：先写死一个索引跑通全链路，再加上动态列，最后补模板和生命周期。一步一步来，比一次性上全量稳得多。环境不一样，遇到的坑也会不一样，但只要这套思路在，排查问题的时候你就知道该往哪个方向使劲。