如何通过Python递归获取Elasticsearch映射的所有字段名与类型

嵌套Mapping全路径字段提取实现 #

直接通过递归遍历properties节点即可提取全路径字段，默认限制最大递归深度为4层，自动拼接点分隔的字段路径，同时兼容object、nested嵌套类型、text字段挂keyword子字段的多字段场景：

from elasticsearch import Elasticsearch

def extract_mapping_fields(es: Elasticsearch, index_name: str, max_depth: int = 4) -> dict:
    """
    递归提取索引mapping全路径字段及对应类型
    返回格式: {"字段全路径": "ES字段类型", 例: "user.addr.location": "geo_point"}
    """
    resp = es.indices.get_mapping(index=index_name)
    root_mapping = resp[index_name]["mappings"]
    field_type_map = {}

    def _dfs(props: dict, current_prefix: str = "", current_depth: int = 1):
        if current_depth > max_depth:
            return
        for field_name, field_conf in props.items():
            full_path = f"{current_prefix}.{field_name}" if current_prefix else field_name
            # 记录当前字段类型
            if "type" in field_conf:
                field_type_map[full_path] = field_conf["type"]
            # 递归遍历嵌套属性（覆盖object、nested类型）
            if "properties" in field_conf:
                _dfs(field_conf["properties"], full_path, current_depth + 1)
            # 处理多字段场景（如text字段挂载的keyword子字段）
            if "fields" in field_conf:
                _dfs(field_conf["fields"], full_path, current_depth + 1)

    if "properties" in root_mapping:
        _dfs(root_mapping["properties"])
    return field_type_map

ES映射类型冲突逻辑说明 #

• 默认dynamic: true模式下，ES仅会对从未出现过的新字段自动新增映射，已存在的字段如果写入不匹配类型的数据，会直接抛出mapper_parsing_exception，不会自动创建同名字段不同类型的映射。
• 两类异常必须在校验环节主动检测，不能依赖ES写入时的自动报错：
  • 索引设置为dynamic: false时，写入的新字段不会进入mapping、不会建索引，仅会存在于_source中，会出现实际数据有字段但mapping无定义的情况
  • 低版本ES中同一路径下object类型与普通字段的写入冲突可能不会第一时间抛错，会导致索引结构损坏，必须提前拦截。

多维度数据校验实现 #

1. 字段类型匹配校验 #

基于前面提取的field_type_map，逐行递归展开待导入数据的全路径字段，按ES类型与Python类型的对应关系做匹配：

• ESlong/integer/short对应Pythonint
• EStext/keyword对应Pythonstr
• ESboolean对应Pythonbool
• ESgeo_point对应包含lat/lon的dict，或格式为"纬度,经度"的字符串
• ESdate对应合法格式时间字符串、毫秒/秒级时间戳
• 类型不匹配时直接记录异常：字段路径、期望类型、实际值、所属数据主键。

2. 空字段检测 #

遍历mapping字段列表，逐行检查字段值是否为None、空字符串、空数组，按业务配置的必填字段列表标记空值异常，注意不要将合法的0、False误判为空值。

3. 字段关联逻辑校验 #

用可配置规则实现，避免硬编码，方便后续扩展校验逻辑，示例配置结构：

# 关联校验规则集
VALIDATION_RULES = [
    {
        "err_msg": "field4为true时geopoint类型field10不能为空",
        "trigger": lambda row: row.get("field4") is True,
        "check": lambda row: row.get("field10") is not None
    },
    {
        "err_msg": "field7为指定值时field8不符合前缀规则",
        "trigger": lambda row: row.get("field7") in [1001, 1003, 1005], # 特定值触发校验
        "check": lambda row: isinstance(row.get("field8"), str) and row["field8"].startswith("biz_") # 自定义校验规则
    }
]

# 校验逻辑
def check_relation_rules(row: dict) -> list:
    errors = []
    for rule in VALIDATION_RULES:
        if rule["trigger"](row) and not rule["check"](row):
            errors.append(rule["err_msg"])
    return errors

你手头准备的索引创建、数据写入、单字段查询示例，可以直接作为单元测试用例：创建测试索引后先提取mapping结构，构造包含正常、异常场景的测试数据，依次跑三类校验逻辑，验证拦截准确率符合预期即可。

内容的提问来源于stack exchange，提问作者Jennifer Crosby