【一起学Python】第92天:NumPy进阶:内存映射、Numba加速与C扩展

import numpy as npimport os# 模拟创建一个 500MB 的二进制数据文件filename = 'large_data.bin'shape = (10_000_000, 10)  # 1000万行 × 10列dtype = np.float32# 写入初始数据（仅演示，实际数据通常来自传感器/日志）if not os.path.exists(filename):    raw = np.random.randn(*shape).astype(dtype)    raw.tofile(filename)  # 连续写入二进制文件# 使用 memmap 映射文件（不占用大量RAM）mmap_arr = np.memmap(filename, dtype=dtype, mode='r+', shape=shape)print(f"文件路径: {filename}")print(f"映射形状: {mmap_arr.shape}")print(f"预估大小: {mmap_arr.nbytes / 1024**3:.2f} GB")print(f"实际内存占用: {mmap_arr.nbytes / 1024**2:.2f} MB (仅元数据)")

# mode 参数说明：# 'r'  : 只读（推荐用于分析）# 'r+' : 读写（修改会直接写入磁盘）# 'w+' : 创建/覆盖写入# 'c'  : Copy-on-Write（修改仅存于内存，不写回磁盘）# 按块读取（避免一次性加载）block_size = 100_000for i in range(0, shape[0], block_size):    chunk = mmap_arr[i : i + block_size, :]  # 仅加载当前块到RAM    chunk_mean = np.mean(chunk, axis=0)    print(f"块 {i//block_size} 均值: {chunk_mean[:3]}...")    del chunk  # 及时释放引用，让OS回收页缓存

import numpy as npfrom numba import njitimport time# 纯 Python 循环（慢）def python_mse(y_true, y_pred):    total = 0.0    for i in range(len(y_true)):        total += (y_true[i] - y_pred[i]) ** 2    return total / len(y_true)# Numba JIT 编译（快）@njitdef numba_mse(y_true, y_pred):    total = 0.0    for i in range(len(y_true)):        total += (y_true[i] - y_pred[i]) ** 2    return total / len(y_true)# 测试数据y_true = np.random.rand(1_000_000)y_pred = np.random.rand(1_000_000)# 首次运行包含编译时间，后续运行极快_ = numba_mse(y_true, y_pred)  # 预热编译start = time.time()result = numba_mse(y_true, y_pred)print(f"Numba耗时: {time.time()-start:.4f}s | 结果: {result:.4f}")# 通常比纯Python快 50~200 倍，接近C语言速度

from numba import prange, vectorize# 并行循环（自动多线程）@njit(parallel=True)def parallel_sum(arr):    total = 0.0    for i in prange(arr.shape[0]):  # prange 替代 range        total += arr[i]    return total# 自定义向量化函数（类似 np.sin）@vectorize(['float64(float64)'])def custom_activation(x):    if x > 0:        return x    else:        return 0.01 * x  # LeakyReLUdata = np.linspace(-5, 5, 1000)activated = custom_activation(data)

# cython_example.pyx (伪代码示例)# distutils: language = c# cython: boundscheck=False, wraparound=Falseimport numpy as npcimport numpy as cnpdef fast_dot(cnp.ndarray[cnp.float64_t, ndim=2] A,              cnp.ndarray[cnp.float64_t, ndim=2] B):    cdef int i, j, k    cdef int m = A.shape[0], n = B.shape[1], p = A.shape[1]    cdef cnp.ndarray[cnp.float64_t, ndim=2] C = np.zeros((m, n))    for i in range(m):        for j in range(n):            for k in range(p):                C[i, j] += A[i, k] * B[k, j]    return C

import numpy as npfrom numba import njit, prangeimport time# 1. 内存映射加载（避免 OOM）filename = 'sensor_log_20gb.bin'shape = (200_000_000, 3)  # 2亿行 × 3列data = np.memmap(filename, dtype=np.float32, mode='r', shape=shape)# 2. Numba 加速滑动窗口计算@njitdef sliding_window_stats(arr, window_size):    n = arr.shape[0]    out_mean = np.empty(n - window_size + 1)    out_std = np.empty(n - window_size + 1)    for i in range(n - window_size + 1):        chunk = arr[i : i + window_size]        out_mean[i] = np.mean(chunk)        out_std[i] = np.std(chunk)    return out_mean, out_std# 3. 分块执行（结合 memmap 与 numba）window = 1000block_rows = 5_000_000results = []for start in range(0, shape[0], block_rows):    end = min(start + block_rows, shape[0])    chunk = data[start:end, 1]  # 仅取第2列温度数据    # Numba 加速计算    m, s = sliding_window_stats(chunk, window)    results.append((m, s))    print(f"处理块 {start//block_rows} 完成")print("超大数据流水线执行完毕！")

from numba import njitimport numpy as np@njitdef weighted_moving_average(arr, weights):    n = len(arr)    w_len = len(weights)    out = np.empty(n - w_len + 1)    for i in range(n - w_len + 1):        s = 0.0        for j in range(w_len):            s += arr[i+j] * weights[j]        out[i] = s / np.sum(weights)    return out# 测试data = np.random.rand(1000)w = np.array([0.1, 0.2, 0.4, 0.2, 0.1])print(weighted_moving_average(data, w)[:5])