如何用Numpy快速转换ConnectorX读取的DataFrame列数据类型？

你当前用str.decode逐列处理的方式，本质是Pandas的逐元素操作，数据量大时效率极低。下面是几种更快的优化方案，重点是基于Numpy的向量化实现：

一、最推荐：从ConnectorX源头避免字节串问题 #

ConnectorX的read_sql支持通过dtype参数提前指定列的数据类型，直接把字符串列解析为Python str，省去后续转换步骤：

import pandas as pd
import connectorx as cx

db_url="mysql://{user}:{pw}@localhost:{port}/{db}".format(user="cookie",pw="cookie123",db="mydb",port=3306)
query='Select * from test'

# 提前定义列类型，把字符串列指定为str
dtype_map = {
    "列名1": str,
    "列名2": str,
    # 其他数值列可按需指定int64等类型
}

# 直接读取为目标类型
iris = cx.read_sql(db_url, query, dtype=dtype_map)

# 批量处理Int64转int64
int_cols = iris.select_dtypes(include=['Int64']).columns
iris[int_cols] = iris[int_cols].astype('int64')

如果不清楚具体列名，可以先读取一次获取列信息再生成dtype_map，或者对所有疑似字符串列统一指定类型。

二、基于Numpy的批量字节串解码（数据已读取后的快速转换） #

用Numpy的np.char.decode做向量化解码，这是底层C级别的批量操作，比Pandas的str.decode快数倍：

import pandas as pd
import connectorx as cx
import numpy as np

db_url="mysql://{user}:{pw}@localhost:{port}/{db}".format(user="cookie",pw="cookie123",db="mydb",port=3306)
query='Select * from test'
iris = cx.read_sql(db_url, query)

# 批量处理所有object类型的字节串列
object_cols = iris.select_dtypes(include=['object']).columns
for col in object_cols:
    # 空值会自动保留，无需额外fillna
    iris[col] = np.char.decode(iris[col].values, encoding='utf-8')

# 批量处理Int64转int64
int_cols = iris.select_dtypes(include=['Int64']).columns
iris[int_cols] = iris[int_cols].astype('int64')

如果部分object列不是字节串，可以加个简单判断：先检查列中是否存在字节串元素，再执行解码。

三、Pandas批量优化（比原方法快，但不如Numpy） #

用apply批量处理所有object列，减少逐元素操作的开销：

object_cols = iris.select_dtypes(include=['object']).columns
iris[object_cols] = iris[object_cols].apply(lambda x: x.str.decode('utf-8').fillna(x))

# 同样批量处理Int64转int64
int_cols = iris.select_dtypes(include=['Int64']).columns
iris[int_cols] = iris[int_cols].astype('int64')

性能对比测试 #

你可以用以下代码直观对比不同方法的速度：

import time

# 原方法计时
start = time.time()
for col, dtype in iris.dtypes.items():
    if dtype == object:
        iris[col] = iris[col].str.decode('utf-8').fillna(iris[col])
    elif dtype == 'Int64':
        iris[col] = iris[col].astype('int64')
print(f"原方法耗时: {time.time() - start:.2f}s")

# Numpy方法计时
start = time.time()
object_cols = iris.select_dtypes(include=['object']).columns
for col in object_cols:
    iris[col] = np.char.decode(iris[col].values, encoding='utf-8')
int_cols = iris.select_dtypes(include=['Int64']).columns
iris[int_cols] = iris[int_cols].astype('int64')
print(f"Numpy方法耗时: {time.time() - start:.2f}s")

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