如何将点分隔字符串转为嵌套数据访问路径以动态获取Elasticsearch数据？

我之前处理Metricbeat数据的时候也碰到过一模一样的问题——硬编码字段路径简直是维护噩梦！别担心，咱们可以通过拆分点分隔字符串+遍历嵌套结构来实现动态取值，完全适配任意这种格式的字段名。

核心思路 #

把"system.cpu.total.pct"这种字符串按.拆分成键的列表["system", "cpu", "total", "pct"]，然后从你的_source数据开始，逐层往下访问每个键，最终拿到目标值。

方案1：基础循环遍历（可读性强） #

写一个通用函数来处理嵌套取值，还能灵活处理路径不存在的情况：

def get_nested_value(data, path_str):
    # 拆分路径为键列表
    keys = path_str.split('.')
    current_data = data
    for key in keys:
        # 检查当前层级是否是字典且包含目标键
        if isinstance(current_data, dict) and key in current_data:
            current_data = current_data[key]
        else:
            # 路径不存在时返回None，也可以根据需求抛出异常
            return None
    return current_data

使用示例

放到你的业务逻辑里就像这样：

# 先拿到固定前缀的数据源
source_data = rawData['hits']['hits'][thisRecord]["_source"]
# 动态传入字段路径
target_value = get_nested_value(source_data, "system.cpu.total.pct")
# 确认有值再添加到数组
if target_value is not None:
    dataArray.append(target_value)

方案2：用reduce简化代码（更简洁） #

如果喜欢更紧凑的写法，可以用functools.reduce配合operator.getitem来一行搞定核心逻辑，异常处理也很简单：

from functools import reduce
import operator

def get_nested_value(data, path_str):
    keys = path_str.split('.')
    try:
        # 逐层访问每个键
        return reduce(operator.getitem, keys, data)
    except (KeyError, TypeError):
        # 键不存在或者中间层级不是可访问结构时返回None
        return None

扩展：处理包含数组索引的路径（可选） #

如果之后你的字段路径里出现数组索引（比如system.cpu.cores.0.usage），可以稍微修改函数来支持：

def get_nested_value(data, path_str):
    keys = path_str.split('.')
    current_data = data
    for key in keys:
        if isinstance(current_data, dict) and key in current_data:
            current_data = current_data[key]
        elif isinstance(current_data, list) and key.isdigit():
            idx = int(key)
            if 0 <= idx < len(current_data):
                current_data = current_data[idx]
            else:
                return None
        else:
            return None
    return current_data

这样不管是纯字典嵌套还是混合数组的结构，都能轻松处理啦！

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