如何高效将Pandas DataFrame指定列外数据转为JSON列？

我有一个大型Pandas DataFrame，包含source_name列。需要生成仅含json_content和source_name两列的输出DataFrame：

• json_content列存储除source_name外所有列转换后的JSON字符串
• 当前用自定义JSON编码器+列表推导实现，但处理速度太慢，求更高效方案
• 已尝试过assign+agg的写法，但还没完成JSON序列化

当前实现代码：

import json
import datetime
import decimal
import numpy as np
import pandas as pd

class CustomJSONEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime.date):
            return obj.isoformat()
        elif isinstance(obj, decimal.Decimal):
            return str(obj)
        elif isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, np.nan):
            return 'None'
        return super().default(obj)

dict_list = df.to_dict(orient='records')
new_dict_list = [
    {'json_content': json.dumps({k: v for k, v in row.items()
                                 if k not in ['source_name']},
                                cls=CustomJSONEncoder),
     'source_name': row['source_name']} for row in
    dict_list]
df = pd.DataFrame(new_dict_list, columns=['json_content',
                                          'source_name'])

尝试过的不完整写法：

df = df[['source_name']].assign(
            json_content=df.iloc[:, 1:].agg(dict, 1))

1. 用Pandas内置to_json（最快方案） #

Pandas的to_json是底层C优化的，比Python层循环+json.dumps快几个数量级。先预处理特殊类型列，再直接按行生成JSON：

import pandas as pd
import datetime
import decimal
import numpy as np

def preprocess_col(col):
    # 处理datetime类型：转ISO格式字符串
    if pd.api.types.is_datetime64_any_dtype(col):
        return col.dt.isoformat()
    # 处理Decimal类型：转字符串
    elif col.apply(lambda x: isinstance(x, decimal.Decimal)).any():
        return col.astype(str)
    # 处理numpy数组：转列表
    elif col.apply(lambda x: isinstance(x, np.ndarray)).any():
        return col.apply(lambda x: x.tolist())
    # 处理NaN：转'None'（和原编码器逻辑一致）
    return col.fillna('None')

# 筛选需要转JSON的列，预处理后生成每行的JSON字符串
json_cols = df.columns.drop('source_name')
df_processed = df[json_cols].apply(preprocess_col, axis=0)
# to_json用lines=True生成每行一个JSON对象，拆分后对应每行数据
df['json_content'] = df_processed.to_json(orient='records', lines=True).split('\n')[:-1]

# 保留目标列
df_result = df[['source_name', 'json_content']]

2. 优化assign+agg写法 #

原agg(dict,1)仅生成字典，补充JSON序列化步骤，利用Pandas内部优化的agg替代手动遍历：

import json

# 复用你的自定义编码器
class CustomJSONEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime.date):
            return obj.isoformat()
        elif isinstance(obj, decimal.Decimal):
            return str(obj)
        elif isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, np.nan):
            return 'None'
        return super().default(obj)

df_result = df[['source_name']].assign(
    json_content=df.drop('source_name', axis=1)
                   .agg(dict, axis=1)
                   .apply(lambda x: json.dumps(x, cls=CustomJSONEncoder))
)

这个写法比原列表推导高效，因为agg(dict)是Pandas内部实现，避免了Python层的逐行字典遍历。

3. 超大型数据集：用swifter自动加速 #

如果DataFrame规模极大，swifter会自动判断是用向量化操作还是多进程apply，进一步提升速度：

import swifter
import json

class CustomJSONEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime.date):
            return obj.isoformat()
        elif isinstance(obj, decimal.Decimal):
            return str(obj)
        elif isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, np.nan):
            return 'None'
        return super().default(obj)

df_result = df[['source_name']].assign(
    json_content=df.drop('source_name', axis=1)
                   .agg(dict, axis=1)
                   .swifter.apply(lambda x: json.dumps(x, cls=CustomJSONEncoder))
)

安装依赖：pip install swifter

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