实现类ClickHouse arrayJoin功能：JSON数组展平为行的算法问题

问题描述 #

需要将嵌套JSON中的数组展开为单独行，同时保留非数组字段的重复值，并将数组元素中的嵌套字段拆分为独立列。示例输入：

{
    "A": "B",
    "C": [{"D": "E"}, {"F": "G"}, "H", {"D": "I"}]
}

期望输出表格：

解决方案 #

核心思路是先遍历JSON结构，收集所有可能的列名（含嵌套路径）并定位需要展开的数组；再针对数组中的每个元素，结合非数组字段的值生成单独行，填充对应列的数据，缺失字段留空。

代码实现 #

def collect_columns_and_arrays(data, current_path=None):
    """收集所有列名，并标记数组路径"""
    if current_path is None:
        current_path = []
    columns = set()
    array_paths = []

    if isinstance(data, dict):
        for key, value in data.items():
            new_path = current_path + [key]
            if isinstance(value, list):
                array_paths.append(new_path)
            sub_cols, sub_arrays = collect_columns_and_arrays(value, new_path)
            columns.update(sub_cols)
            array_paths.extend(sub_arrays)
    elif isinstance(data, list):
        # 数组本身不生成列，列来自数组元素
        for item in data:
            sub_cols, sub_arrays = collect_columns_and_arrays(item, current_path)
            columns.update(sub_cols)
            array_paths.extend(sub_arrays)
    else:
        # 非嵌套值，生成列名（用点连接路径）
        column_name = ".".join(current_path)
        columns.add(column_name)
    
    return columns, array_paths

def flatten_array(data, array_path, parent_data=None):
    """展开指定路径的数组，生成多行数据"""
    if parent_data is None:
        parent_data = {}
    
    # 获取数组数据
    array_data = data
    for key in array_path:
        array_data = array_data[key]
    
    # 获取非数组部分的字段值
    non_array_data = {}
    def extract_non_array(d, path=None):
        if path is None:
            path = []
        if isinstance(d, dict):
            for k, v in d.items():
                new_path = path + [k]
                if new_path != array_path and not isinstance(v, list):
                    if isinstance(v, dict):
                        extract_non_array(v, new_path)
                    else:
                        col_name = ".".join(new_path)
                        non_array_data[col_name] = v
                elif new_path == array_path:
                    continue
                else:
                    # 其他数组暂时不处理（若有多层数组可扩展）
                    pass
    extract_non_array(data)
    
    # 处理数组中的每个元素，生成行
    rows = []
    for item in array_data:
        row = non_array_data.copy()
        # 处理数组元素，填充对应列
        def fill_row(item, path=None):
            if path is None:
                path = array_path
            if isinstance(item, dict):
                for k, v in item.items():
                    new_path = path + [k]
                    fill_row(v, new_path)
            elif isinstance(item, list):
                # 这里示例中没有多层数组，若有可递归处理
                pass
            else:
                col_name = ".".join(path)
                row[col_name] = item
                # 如果是数组的直接值（非对象），填充数组本身的列
                if len(path) == len(array_path):
                    row[".".join(array_path)] = item
        fill_row(item)
        rows.append(row)
    
    return rows

def array_join_json(data):
    """主函数：处理JSON，生成展开后的行和列"""
    columns, array_paths = collect_columns_and_arrays(data)
    # 假设只有一个顶层数组（若有多个可扩展处理）
    if not array_paths:
        return [data], columns
    
    # 取第一个数组路径（示例中是["C"]）
    array_path = array_paths[0]
    rows = flatten_array(data, array_path)
    
    # 确保所有列都在每行中存在，缺失的设为空字符串
    full_columns = sorted(columns)
    formatted_rows = []
    for row in rows:
        formatted_row = {col: row.get(col, "") for col in full_columns}
        formatted_rows.append(formatted_row)
    
    return formatted_rows, full_columns

# 示例使用
if __name__ == "__main__":
    input_json = {
        "A": "B",
        "C": [{"D": "E"}, {"F": "G"}, "H", {"D": "I"}]
    }
    
    rows, columns = array_join_json(input_json)
    
    # 打印表格表头
    print("| " + " | ".join(columns) + " |")
    print("| " + " | ".join(["---"]*len(columns)) + " |")
    # 打印每行数据
    for row in rows:
        print("| " + " | ".join(str(row[col]) for col in columns) + " |")

代码解释 #

1. collect_columns_and_arrays：遍历JSON结构，收集所有嵌套路径对应的列名，同时记录需要展开的数组路径（如示例中的["C"]）。
2. flatten_array：提取指定路径的数组元素，保留非数组字段的重复值；逐个处理数组元素，将简单值填充到数组对应列，将对象嵌套值填充到对应嵌套列。
3. array_join_json：整合上述逻辑，生成包含所有列的完整行数据，缺失值自动填充为空字符串。

运行示例代码后，会输出符合要求的表格格式结果。

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