Spark Structured Streaming 动态写入 Elasticsearch Index 完整实践

如果你做过实时数仓大概率会遇到这个场景Kafka 里接入的日志或业务数据需要按业务类型、按天、甚至按租户写进不同的 Elasticsearch 索引。刚开始偷懒把 index 写死成一个常量等需求一变——A 业务数据生命周期 7 天B 业务数据要留 3 个月你就得被迫把动态写入 ES Index这件事提上日程。我最早接触Spark Structured Streaming时也天真地以为动态 index 只是把logs换成slogs-${date}这么简单。实际做下来发现里面牵扯到连接器选型、批次语义、ES 索引模板、幂等写入、数据倾斜调优一堆问题。这篇文章把我实现Spark Structured Streaming 基于数据动态写入 ES Index的完整思路和踩坑记录整理出来适合正在做实时数据入 ES、或者准备把写死的 index 改成动态方案的人参考。1. 为什么Index 写死这件事会逼着你重构1.1 那些开始以为不重要的业务约束先说场景。之前接手过一个数据平台需求数据流很简单Kafka 里一个 topic里面有不同业务类型的日志统一进 ES 做检索和分析。第一版我图省事所有数据写进一个logs-common索引加个logType字段区分。上线半个月问题就来了运维要按业务类型做索引生命周期管理A 业务数据只保留 3 天B 业务要保留 180 天一个索引没法同时满足。某个业务量暴涨全集群的分片压力都被它带起来其他业务查询被拖慢。清理旧数据时只能按时间删文档删起来又慢又占资源远不如直接删索引干净。这种情况下物理上拆分索引几乎成了必选项。而拆分索引的规则往往来自数据本身日志里带logType、tenantId、date每条数据该进哪个索引必须根据这条数据的内容现场算出来。这就是动态写入的起点。1.2 动态到底在动什么所谓动态写入 ES Index本质上是把indexName从一个写死的常量变成一个可以由行数据推导出的变量。一般由三部分拼接而来业务维度logType、tenantId、bizType时间维度yyyy-MM-dd、yyyy-MM用于滚动索引前缀规范统一以logs-、app-开头方便索引模板匹配举几个真实常见的组合logs-order-2024-06-01按业务类型 天切分audit-tenant-8823按租户维度隔离metrics-server-2024-06按月度聚合运维指标动态不等于乱动。设计的时候要保证同一个业务语义的数据在同一个索引下索引名是可预测、可治理的而不是每条数据的 index 都随机生成否则 ES 集群会被几百个小索引的分片拖垮。1.3 什么时候我劝你不要用动态索引动态索引不是银弹。我后来复盘发现很多场景其实不需要动态切索引硬上反而增加运维压力。数据量本身不大单索引一天几 GB 以内查询也基本是全量检索那一个索引 字段过滤就够没必要拆。业务经常做跨索引聚合比如同时算 A 业务和 B 业务的汇总指标。拆成多个索引后ES 跨索引查询要么用index pattern要么建别名复杂度比单索引高得多。动态索引数量失去控制。索引数量一多每个索引至少一个分片分片本身会占用堆内存和文件句柄。几百个索引的分片开销会直接压垮小集群。判断标准很简单如果有明确的生命周期差异、隔离需求或容量规划需求动态索引是合理的如果只是觉得动态看起来很灵活我建议忍住。2. 版本匹配与依赖搭设先让 Spark 和 ES 握手成功2.1 版本矩阵是第一道坑Spark 和 ES 的版本兼容性直接决定了你的代码能跑得多顺。ES 8.x 之后RestHighLevelClient逐渐被新客户端替代而elasticsearch-spark连接器版本又必须和 Spark 大版本、Scala 版本双匹配。这一步确实容易卡住新手。Spark 版本Scala 版本推荐连接器说明Spark 3.3.x2.12elasticsearch-spark-30_2.127.17.x最稳妥的组合Spark 3.4.x2.12 / 2.13elasticsearch-spark-30_2.12或_2.13注意连接器同样要换 Scala 后缀Spark 3.5.x2.12 / 2.13同上8.x 连接器需单独验证优先用 7.17.x 连接器测通再迁移ES 8.x 集群任意连接器 8.x 新版 Java Client代码 API 有较大变化我用的组合是 Spark 3.3.2 Scala 2.12 ES 7.17 集群。如果你们 ES 已经是 8.x建议先在测试环境用小数据量验证连接器行为尤其是es.mapping.id、es.write.operation这些参数在 8.x 下有没有变化。2.2 依赖配置参考build.sbt里加上这些依赖就能开始libraryDependencies Seq( org.apache.spark %% spark-sql % 3.3.2, org.apache.spark %% spark-sql-kafka-0-10 % 3.3.2, org.elasticsearch %% elasticsearch-spark-30 % 7.17.9, org.elasticsearch.client % elasticsearch-rest-high-level-client % 7.17.9, org.apache.kafka %% kafka % 3.3.0 )注意elasticsearch-spark-30_2.12在 Maven 里的坐标写法是org.elasticsearch:elasticsearch-spark-30_2.12:7.17.9。这里直接用它做流式写入同时引入rest-high-level-client是为了在代码里灵活管理索引创建和手动 bulk 写入。如果你连 Kafka 也不打算接可以把spark-sql-kafka-0-10去掉直接从一个本地 socket 流测试。2.3 Schema 设计是后面所有代码的地基动态 index 的解析逻辑要稳定前提是输入数据里有可靠的路由字段。我建议在接入层就统一好 schema字段类型用途_idstringES 文档主键幂等写的核心logTypestring业务类型决定 index 前缀daystring日期字符串格式yyyy-MM-ddpayloadmap实际业务数据timestamplong事件时间用于乱序处理为什么day不直接用 Spark 的处理时间因为流处理存在延迟和重跑处理时间不等于事件时间用数据本身的业务日期来切分索引才符合业务预期。这些字段最好在from_json解析时统一提取出来后面的resolveIndexName函数就能写得很干净。3. 核心实现foreachBatch 里让 IndexName活起来3.1 为什么不能直接用 writeStream 写 ESStructured Streaming 的writeStream虽然支持输出到 ES 连接器但它的资源名基本是启动时固定的做不到每行数据动态决定 index。如果用foreach这种逐行输出方式每条数据单独发一次 bulk性能直接被打爆。正确做法是用foreachBatch。每批微批数据会以一份完整 DataFrame 的形式交给你的函数这时候你可以做聚合、过滤、重分区再按自定义逻辑把数据写进不同的 ES 索引。相当于把流式问题变成微批批处理问题Structured Streaming 依然负责 checkpoint 和 offset 管理你只需要稳扎稳打处理每一批。3.2 动态构造 IndexName 的解析函数路由逻辑收敛到一个函数里方便测试和复用def resolveIndexName(row: Row): String { val logType row.getAs[String](logType) val day row.getAs[String](day) slogs-$logType-$day }这里有几个细节index 统一小写字母开头ES 索引名不允许大写。day在写入前就格式化好不要在函数里临时格式化避免不同时区导致索引分裂。如果路由字段缺失宁可抛异常让任务失败也不要写入一个logs-null的脏索引。3.3 用 foreachBatch 分区内聚批量写 ES核心代码框架如下val esOptions Map( es.nodes - es-node-1:9200,es-node-2:9200, es.nodes.wan.only - true ) kafkaDF .selectExpr(CAST(value AS STRING) as json) .select(from_json(col(json), mySchema).as(data)) .select(data.*) .writeStream .option(checkpointLocation, /tmp/ckpt/ckpt_dynamic_index) .foreachBatch { (batchDF: DataFrame, batchId: Long) // 缓存这批数据避免后面多次遍历重新计算 batchDF.persist(StorageLevel.MEMORY_AND_DISK) // 关键优化按 indexName 重分区让同一个索引的数据尽量进入同一个 Spark 分区 val indexedDF batchDF.withColumn(indexName, expr(concat(logs-, logType, -, day))) indexedDF .repartition(48, col(indexName)) .foreachPartition { iter val client EsClientHolder.getClient() val bulkBatchSize 1000 // 在一个分区内先按 indexName 分组 iter .map(row row.getAs[String](indexName) - row) .toArray .groupBy(_._1) .foreach { case (indexName, rows) ensureIndexExists(client, indexName) rows.grouped(bulkBatchSize).foreach { group val bulkRequest new BulkRequest() group.foreach { case (_, row) bulkRequest.add( new IndexRequest(indexName) .id(row.getAs[String](_id)) .source(convertRowToJson(row), XContentType.JSON) ) } val response client.bulk(bulkRequest, RequestOptions.DEFAULT) if (response.hasFailures) { // 失败时要打日志并抛出异常让 Spark 任务失败后重试 val failures response.getItems.filter(_.isFailed) throw new RuntimeException(s写ES失败: ${failures.mkString(,)}) } } } } batchDF.unpersist() } .start() .awaitTermination()convertRowToJson就是把 Row 转成 JSON 字符串的辅助函数用 jackson 或者 Spark 内置的row.json都可以。EsClientHolder是一个简单的单例维护一个 ES 高可用客户端避免每个批次都创建连接。这段代码的核心思路是先在 driver 端拿到每个微批的 DataFrame算出indexName列再通过repartition按该列做数据重分布。executor 端在每个 partition 内把同一 index 的数据聚合起来用 bulk 批量写入。这样既实现了动态 index又避免了写一行发一次请求的低效做法。3.4 ES 索引不存在与模板管理动态写入时ES 不会因为你写入一个新 index 就自动给你建好合理配置的索引。默认情况下可能自动创建出单分片、单副本、动态 mapping 的索引这在生产环境是不能接受的。更优雅的方案是用 ES Index Template让匹配logs-*的索引自动套用模板PUT _index_template/logs_template { index_patterns: [logs-*], template: { settings: { number_of_shards: 3, number_of_replicas: 1, refresh_interval: 10s }, mappings: { dynamic_templates: [ { strings_as_keyword: { match_mapping_type: string, mapping: { type: keyword } } } ], properties: { timestamp: { type: date }, _id: { type: keyword } } } }, priority: 100 }有了模板写入logs-order-2024-06-01时如果索引不存在ES 会自动按模板创建。如果你们的 ES 版本较老7.8 以下用 PUT_template旧模板 API 即可。代码里保留一个ensureIndexExists作为兜底但大多数场景都被模板接管了。提示动态索引数量一定要有上限管控。模板让创建索引零成本但分片数不会变几百个动态索引就是几百个分片。我会在监控里加一个索引数量超过阈值的告警及时发现问题。3.5 _id 与幂等写入设计动态 index 可以做到但同一批数据因故障重跑带来的重复写问题才是真正的坑。Structured Streaming 的 checkpoint 能保证 Spark 侧的处理进度不丢但它无法保证 ES 侧写入是原子的。同一批数据在故障恢复后可能被再次执行如果写入时没有幂等设计ES 里就会出现重复文档。我在项目中走了两条路对于状态类数据如订单、用户档案用业务主键作为_id写入走upsert语义。ES 文档 id 相同会覆盖旧文档天然幂等。对于日志类数据如访问日志Kafka 消息本身没有稳定 key我为每条日志生成一个md5(时间戳 日志内容片段 随机数)作为_id。重复消费时只要内容不变_id就能复现写入直接覆盖。第二种方案有个前提相同的日志被重复消费时内容完全一致。如果中间做了数据清洗或字段补充_id的计算要在清洗之前的原始数据上完成否则每次重跑都可能算出不同 id。4. 上线才暴露的坑事务假象、字段冲突与写入拒绝4.1 微批 Sink 不是你以为的事务先说一个我在项目中踩得最深的坑foreachBatch这个函数名太容易让人误以为它有一个事务边界。实际上它只是一个回调函数每批微批数据到达时被调用一次。如果写 ES 中途发生部分成功、部分失败Spark 只能整体标记这一批失败并重跑而 ES 里已经写入的那部分文档并不会自动回滚。有一个直观的经验是把每次微批的数据切到尽量小让重跑代价可控。宁可多跑几批也不要一批里面塞几百万条失败后重跑一次要等很久。我在压测时遇到过单批太大导致 ES 节点 OOM整个流作业重启然后再跑一批又 OOM形成一个恶性循环。4.2 乱序数据覆盖新数据Kafka 的多个分区不能保证数据全局有序而 Spark 的微批处理又会把不同分区数据重新洗牌。如果业务里存在先到的旧状态覆盖后到的新状态这种问题ES 的 upsert 语义反而会放大乱序问题。解决方案是在写入前先按_id做一次微批内的去重排序batchDF .withColumn(ts, col(timestamp).cast(long)) .groupBy(_id) .agg(max(ts).as(max_ts)) .join(batchDF, Seq(_id)) .filter(col(timestamp).cast(long) col(max_ts)) .drop(max_ts)这个操作只解决一个微批内的乱序跨批次的乱序比如延迟数据晚到很久还得靠业务容忍或者在上游 Kafka 端做分区键设计。如果你们的消息 key 设计得好同一实体的数据本身就会进同一个 Kafka 分区乱序问题会小很多。4.3 index 不存在、mapping 冲突、refresh 不可见三连坑上线第四天遇到的三个问题值得单独记一笔。第一个是IndexNotFoundException。原因是我在测试环境关了action.auto_create_index而索引模板还没建好动态 index 写入直接报错。解决方案是先配好模板再启动流任务。第二个是查询用term查某个字段突然报类型错误。排查之后发现同一个字段在logs-order-*索引里被动态映射成了text在另一个索引里被映射成了keyword导致跨索引查询时字段类型冲突。所以模板里用dynamic_templates把所有 string 字段统一映射为keyword和text双字段或者干脆显式定义 properties。这一步最好在索引模板里提前定死不要指望动态映射。第三个最有迷惑性数据明明写成功了但查不到。ES 的写入是近实时的refresh_interval决定了文档从可见到不可见的延迟。如果批量导入阶段把refresh_interval调成了-1禁用 refresh查询就会迟迟看不到数据。这种设置在生产导入期间是有效的加速手段但弄完一定要恢复否则业务方会一直以为数据丢了。4.4 bulk 请求被拒绝的完整排查链路ES 集群在写入压力大时会直接拒绝 bulk 请求报错形如es_rejected_execution_exception。我第一次遇到时只看到 Spark 侧批量失败重试却找不到根因。后来按这个链路排查第一步看 ES 节点线程池状态ThreadPoolExecutor里的write线程池是否打满。线程池满意味着写入请求积压严重。第二步看 ES 日志里的 rejected 堆栈确认是被write线程池拒绝还是被search线程池拒绝。搜索压力大也可能挤压写入线程资源。第三步看 Spark 侧的写入速率和批次大小。如果每批数据量大且分区数多同一时间到达 ES 的 bulk 请求数会非常高即使每个 bulk 只有少量数据也可能打爆线程池。第四步对症下药降低批次分区数、加大 bulk 批次条数让单次 bulk 更大但数量更少、适当调大 ES 节点堆内存、增加节点数。这个链路是整个项目里最耗时的一次排查但理顺之后我意识到大部分动态 index 写入失败的问题根源都不在动态逻辑而在写入速率和集群容量不匹配。5. 把动态写入从能跑调成稳定跑5.1 批次大小和速率的节奏控制Structured Streaming 自带背压和限流参数但不少人会在上线时忽略它们。我习惯这样起步kafkaDF.writeStream .option(maxOffsetsPerTrigger, 200000) .option(spark.sql.shuffle.partitions, 48) ...maxOffsetsPerTrigger限制每个批次最多消费的 Kafka offset 数生产环境可以先设为 10 万到 20 万观察处理速度然后逐步往上调直到 ES 侧出现压力。spark.sql.shuffle.partitions控制后续 shuffle 和 join 的默认分区数这个值太小会导致单个 executor 处理太重太大会导致给 ES 的请求碎片化。48 或 96 是常见起步区间具体要看任务所在机器核心数。5.2 分区内聚与数据倾斜的取舍前面提到我用repartition(48, col(indexName))做内聚这是把双刃剑好处同一个 ES index 的数据尽量落在同一个 Spark 分区bulk 请求更集中连接复用率更高。坏处热门 index 的数据量远大于冷门 index导致少数几个 executor 负载极高出现数据倾斜。我的折中方案是根据预估数据量选一个合理的分区数比如 48 个分区写 20 个动态索引让每个索引平均分到 2-3 个分区。不要让repartition的并行度就等于索引数留出冗余能有效缓解单索引热度过高的问题。如果业务数据真的出现一个巨型 index也有一个降级思路把该 index 的数据额外加一个随机后缀做二次拆分比如logs-hot-2024-06-01-1、logs-hot-2024-06-01-2再通过 ES 别名统一查询。这个方案能解决倾斜但需要业务方接受物理索引拆成多片的治理方式。5.3 ES 侧调优备忘参数默认值导入期建议备注refresh_interval1s30s 或-1导入完成要恢复number_of_replicas10大批量写入时可以关副本记得恢复translog durabilityrequestasync允许小概率丢数据时使用index buffer size10% 堆适当调大提升索引写入吞吐这些参数大多是索引级别的如果索引由模板动态创建直接在模板里设置默认值新索引就会自动继承。5.4 监控层面盯四个数动态写入的项目比静态写入更需要监控因为你不知道下一秒会冒出多少个新索引。活跃索引数量每天应该新增几个索引突然多了几十个说明路由规则出了问题。分片数量与节点磁盘水位动态索引激增会直接推高分片数。Spark 流作业处理延迟这个值持续上涨说明作业跟不上数据产生速度。ES bulk rejection 次数有拒绝不要慌关键看是否持续发生持续发生就要降速或扩容。我在监控面板里加了一个简单的索引数量查询GET _cat/indices/logs-*的输出行数超过阈值就报警。这个比任何间接指标都直观。最后再分享一点个人的体会动态写入 ES Index 这件事最大的坑往往不在技术代码而在你到底需要多少个索引、每个索引怎么管理。如果一开始就把索引模板、分片规划、_id策略、限流参数这些定清楚后面运行起来会很省心。尤其对索引数量一定要有敬畏心它不是越多越灵活而是越少越好治理。真正需要拆的时候按业务类型和时间维度拆到恰到好处就已经是一个很成熟的实时数仓方案了。