Spark写入Elasticsearch时日期字段解析失败，如何修复？

问题重现 #

错误信息：

Caused by: org.apache.spark.util.TaskCompletionListenerException: 批量操作[1/1]无法写入所有条目。错误示例（前5条错误信息）：
org.elasticsearch.hadoop.rest.EsHadoopRemoteException: mapper_parsing_exception: 无法解析ID为'BcivPIkBruqga4Hh9x2u'的文档中类型为[date]的字段[user_first_updated]。字段值预览：'2023-07-09 17:54:01.761';org.elasticsearch.hadoop.rest.EsHadoopRemoteException: illegal_argument_exception: 无法使用格式[strict_date_optional_time||epoch_millis||yyyy-MM-dd HH:mm:ss.SSZ]解析日期字段[2023-07-09 17:54:01.761];org.elasticsearch.hadoop.rest.EsHadoopRemoteException: date_time_parse_exception: 所有内置解析器均解析失败
{"index":{}}
{"tenant_id":"KAR1234","username":"bsqlcfg0","learned":false,"user_first_updated":"2023-07-09 17:54:01.761","user_last_updated":"2023-07-09 22:00:00.000","target_table":"user_learning_stats"}

当前Elasticsearch映射配置：

"user_first_updated": {
  "format": "strict_date_optional_time||epoch_millis||yyyy-MM-dd HH:mm:ss.SSZ",
  "type": "date"
}

问题原因 #

映射中user_first_updated字段的格式要求日期末尾带时区标识（Z代表UTC时区），但实际传入的日期值2023-07-09 17:54:01.761没有时区信息，导致Elasticsearch无法匹配解析格式，抛出解析异常。

修复方法 #

方法一：修改数据生成端，补充时区信息 #

在数据处理环节（如Spark），将日期格式化为带时区标识的格式，确保和映射要求匹配：

• 示例Spark代码（Scala）：

import org.apache.spark.sql.functions.date_format
import org.apache.spark.sql.Column

// 将日期字段格式化为带UTC时区标识的格式
val processedDf = originalDf.withColumn(
  "user_first_updated",
  date_format(col("user_first_updated"), "yyyy-MM-dd HH:mm:ss.SSS'Z'")
)

• 若使用其他时区，可替换为对应偏移格式，比如东八区可格式化为yyyy-MM-dd HH:mm:ss.SSS+08:00。

方法二：更新Elasticsearch映射，兼容无时区日期 #

修改字段的格式配置，添加无时区的日期格式规则，让Elasticsearch能解析传入的日期值：

1. 执行映射更新命令（通过Kibana Dev Tools或curl）：

PUT /your_index_name/_mapping
{
  "properties": {
    "user_first_updated": {
      "format": "strict_date_optional_time||epoch_millis||yyyy-MM-dd HH:mm:ss.SSZ||yyyy-MM-dd HH:mm:ss.SSS",
      "type": "date"
    }
  }
}

2. 替换命令中的your_index_name为实际索引名称即可。

方案选择 #

• 若业务要求日期带时区信息，优先选择方法一，保证数据的时区一致性；
• 若数据本身无需时区，或无法修改数据生成逻辑，选择方法二更快捷。

内容的提问来源于stack exchange，提问作者Karthik Sana