如何高效将超大规模CSV文件转换为NPY格式？

你尝试的代码仅适用于小型文件：

import numpy as np

filename = "myfile.csv"
vec =np.loadtxt(filename, delimiter=",")
np.save(f"{filename}.npy", vec)

但处理1200万行×1024列的大规模CSV时，全量加载会占用巨量内存，以下针对你的三个问题给出具体解决方案：

问题1：是否存在针对大型CSV文件的高效转NPY格式的方法？ #

有，核心思路是预分配磁盘映射内存+分块写入，无需全量加载数据到内存。NPY是连续二进制格式，用np.memmap直接操作磁盘空间，完美匹配这种特性：

import numpy as np
import pandas as pd

csv_path = "myfile.csv"
npy_path = "myfile.npy"

# 1. 获取CSV的总行数和列数
with open(csv_path, 'r') as f:
    first_line = f.readline()
    cols = len(first_line.strip().split(','))
    total_rows = sum(1 for _ in f)

# 2. 预分配磁盘映射数组（可根据实际数据类型调整dtype）
dtype = np.float32
memmap_arr = np.memmap(npy_path, dtype=dtype, mode='w+', shape=(total_rows, cols))

# 3. 分块读取CSV并写入磁盘映射
chunksize = 100000  # 每次读取10万行，可根据内存情况调整
for i, chunk in enumerate(pd.read_csv(csv_path, chunksize=chunksize, header=None)):
    start_idx = i * chunksize
    end_idx = start_idx + len(chunk)
    memmap_arr[start_idx:end_idx] = chunk.values.astype(dtype)

# 释放引用，确保数据写入磁盘
del memmap_arr

问题2：若没有直接高效转换的方法，是否可高效迭代读取CSV并写入NPY？ #

可以，但绝对禁止用np.vstack——每次调用都会重新分配内存并复制数据，效率极低。推荐两种高效方式：

方式1：继续用np.memmap预分配后逐块写入（同问题1的方案） #

方式2：分块保存小NPY后合并 #

适合无法提前统计总行数的场景，合并时仍依赖np.memmap避免全量加载：

import numpy as np
import pandas as pd
import os

csv_path = "myfile.csv"
temp_dir = "temp_npy_chunks"
os.makedirs(temp_dir, exist_ok=True)

chunksize = 100000
dtype = np.float32

# 1. 分块保存为小NPY文件
for i, chunk in enumerate(pd.read_csv(csv_path, chunksize=chunksize, header=None)):
    chunk_arr = chunk.values.astype(dtype)
    np.save(os.path.join(temp_dir, f"chunk_{i}.npy"), chunk_arr)

# 2. 统计所有分块的总形状
chunk_files = sorted(os.listdir(temp_dir))
total_rows = 0
cols = None
for f in chunk_files:
    arr = np.load(os.path.join(temp_dir, f), mmap_mode='r')
    total_rows += arr.shape[0]
    cols = cols or arr.shape[1]

# 3. 预分配磁盘映射合并分块
merged_arr = np.memmap("myfile.npy", dtype=dtype, mode='w+', shape=(total_rows, cols))
current_idx = 0
for f in chunk_files:
    arr = np.load(os.path.join(temp_dir, f), mmap_mode='r')
    merged_arr[current_idx:current_idx+arr.shape[0]] = arr
    current_idx += arr.shape[0]

del merged_arr

# 清理临时文件
for f in chunk_files:
    os.remove(os.path.join(temp_dir, f))
os.rmdir(temp_dir)

问题3：若前两种方案不可行，是否有其他高效存储格式（如tensorstore）可存储并高效转为numpy数组？ #

推荐两种方案：

方案1：Zarr（支持动态扩展维度，无需提前指定大小） #

Zarr是专为大规模数组设计的格式，支持按需读取，转为numpy数组几乎无开销：

import zarr
import pandas as pd
import numpy as np

csv_path = "myfile.csv"
zarr_path = "myfile.zarr"

# 获取列数
with open(csv_path, 'r') as f:
    first_line = f.readline()
    cols = len(first_line.strip().split(','))

# 创建可动态扩展的Zarr数组
dtype = np.float32
zarr_arr = zarr.open(zarr_path, mode='w', shape=(0, cols), dtype=dtype, chunks=(100000, cols))

# 分块读取CSV并追加
chunksize = 100000
for chunk in pd.read_csv(csv_path, chunksize=chunksize, header=None):
    chunk_arr = chunk.values.astype(dtype)
    zarr_arr.append(chunk_arr)

# 转为numpy数组（按需读取，无需全量加载）
full_np_arr = zarr_arr[:]  # 读取全部数据
partial_np_arr = zarr_arr[1000:2000]  # 读取部分数据

方案2：TensorStore（需提前获取维度，适合超大规模数据） #

TensorStore需要提前指定维度，但可以先遍历CSV获取总行数和列数再创建：

import tensorstore as ts
import pandas as pd
import numpy as np

csv_path = "myfile.csv"
ts_path = "myfile_ts"

# 获取总行数和列数
with open(csv_path, 'r') as f:
    first_line = f.readline()
    cols = len(first_line.strip().split(','))
    total_rows = sum(1 for _ in f)

# 创建TensorStore
spec = {
    "driver": "n5",
    "kvstore": {"driver": "file", "path": ts_path},
    "metadata": {
        "dimensions": [total_rows, cols],
        "dtype": "float32",
        "blockSize": [100000, cols]
    }
}

ts_arr = ts.open(spec, create=True).result()

# 分块写入
chunksize = 100000
for i, chunk in enumerate(pd.read_csv(csv_path, chunksize=chunksize, header=None)):
    start = i * chunksize
    end = start + len(chunk)
    chunk_arr = chunk.values.astype(np.float32)
    ts_arr[start:end, :] = chunk_arr

# 转为numpy数组
full_np_arr = ts_arr[:].read().result()

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