如何在ElasticSearch创建索引并推送流数据？Spark Elastic Sink异常求助

嘿，我来帮你搞定Elasticsearch流数据推送的问题，先从基础操作讲起，再针对性排查你遇到的诡异情况～

一、先搞懂基础：如何创建ES索引并推送流数据 #

1. 创建ES索引的两种方式 #

• 手动创建（推荐，可控性拉满）：用REST请求自定义索引的分片、副本和字段映射，比如：

  curl -X PUT "localhost:9200/your_index_name" -H 'Content-Type: application/json' -d'
  {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 0
    },
    "mappings": {
      "properties": {
        "id": {"type": "integer"},
        "name": {"type": "text"},
        "create_time": {"type": "date", "format": "yyyy-MM-dd HH:mm:ss"}
      }
    }
  }'
  

• 自动创建：让Spark ES连接器帮你自动生成索引，但前提是ES开启了action.auto_create_index（默认是开的，除非集群改过配置）

2. Spark流推送数据的基本流程 #

首先要确保你引入了版本完全匹配的ES Spark连接器依赖，比如ES是7.17.x，就用这个Maven依赖：

<dependency>
  <groupId>org.elasticsearch</groupId>
  <artifactId>elasticsearch-spark-20_2.12</artifactId>
  <version>7.17.0</version>
</dependency>

之后就是写流查询的代码，你的框架没问题，但细节需要调整。

二、解决你的核心问题：偶尔创建索引但不推数据/连索引都建不了 #

你的代码没报错但不干活，大概率是配置细节踩坑了，我帮你逐一排查：

1. 版本不兼容（最容易踩的坑！） #

Spark、ES服务器、ES Spark连接器这三者的版本必须严格对应！比如你用ES 8.x，就不能用7.x的连接器，反之亦然。版本不匹配会导致各种“无报错但完全没效果”的诡异情况，先确认这一点，绝对是首要排查项。

2. 废弃的参数导致索引异常 #

你用的es.resource参数在ES 7.x及以后已经被废弃了！因为ES从7.x开始移除了type的概念，所以应该用新的参数es.write.resource，格式直接写索引名就行（不需要加type）：

.option("es.write.resource", "your_target_index")

继续用旧参数的话，在高版本ES下会导致索引创建失败或者数据无法写入。

3. ES自动创建索引的开关被关了 #

检查ES集群的action.auto_create_index配置，默认是true，但如果管理员改过这个值为false，Spark就没法自动创建索引。你可以用这条命令查看：

curl -X GET "localhost:9200/_cluster/settings?include_defaults=true&pretty" | grep "action.auto_create_index"

如果是关闭的，要么手动提前创建索引，要么临时开启允许创建你的目标索引：

curl -X PUT "localhost:9200/_cluster/settings" -H 'Content-Type: application/json' -d'
{
  "persistent": {
    "action.auto_create_index": "your_target_index"
  }
}'

4. Checkpoint目录用了临时目录 #

你用的/tmp/是系统临时目录，机器重启或者Spark进程重启后，这个目录很可能被清理，导致流查询的状态丢失，进而出现数据写入异常。建议换成一个永久的、Spark进程有读写权限的目录，比如：

.option("checkpointLocation", "/opt/spark/checkpoint/es_stream_sink/")

5. 数据源根本没数据流入 #

有时候问题不在ES sink，而是你的流数据源根本没输出数据！你可以先加个console sink测试一下：

df.writeStream
  .outputMode("append")
  .format("console")
  .start()
  .awaitTermination()

看看控制台有没有打印数据，如果没有，那得先排查数据源的问题。

6. 数据类型和ES映射不兼容 #

如果你的DataFrame里有ES无法识别的复杂类型（比如自定义嵌套结构、特殊格式的日期），可能会导致写入失败但不抛出异常（Spark流的容错机制会吞掉这类错误）。这时候建议手动创建索引并指定好匹配的映射，再运行流查询。

7. 给你一个修正后的完整代码示例（适配ES 7.x+） #

df.writeStream
  .outputMode("append")
  .format("org.elasticsearch.spark.sql")
  .option("es.nodes", "localhost:9200")
  .option("es.index.auto.create", "true") // 显式开启自动创建索引
  .option("es.write.resource", "your_target_index") // 替换废弃的es.resource
  .option("checkpointLocation", "/opt/spark/checkpoint/es_stream_sink/") // 永久checkpoint目录
  .option("es.batch.size.entries", "1000") // 可选：调整批次写入的大小，优化性能
  .start()
  .awaitTermination()

最后给你一套排查流程，按顺序来准没错 #

1. 先确认Spark、ES、ES连接器的版本完全匹配
2. 把es.resource换成es.write.resource（ES 7.x+）
3. 检查ES的自动创建索引开关，或者手动创建好索引
4. 更换checkpoint目录为永久目录
5. 用console sink验证数据源是否有数据
6. 确保DataFrame的字段类型和ES索引映射匹配

内容的提问来源于stack exchange，提问作者sandeep reddy