企业官网建设流程全解析

简介：本资源是一套面向算法工程师、高校研究者及时间序列分析学习者的多变量时间序列异常检测基准数据集整合包，聚焦工业监控、航天遥测、水处理等关键场景的异常识别需求，解决初学者难获取规范数据、研究人员苦于预处理耗时的共性痛点。压缩包共10个文件（4个CSV原始/处理后数据集、2个TXT说明文档、1个核心Python预处理脚本process_datasets.py、1个HTML索引页及.gitignore、.inscode配置文件），总大小30MB，结构清晰，开箱即用。已有524人学习下载，体现较强实践认可度。用户可直接运行提供的Python脚本完成SMD、SMAP/MSL、SWaT、WADI四大主流数据集的统一时间戳对齐、标签标准化与格式转换；配套processing_summary.txt详述各数据集字段含义与异常标注逻辑，index.html提供可视化导航，大幅降低复现实验门槛，加速模型验证与算法迭代进程。

1. 多变量时间序列异常检测数据集整理：为什么你跑不通别人的代码，大概率卡在数据这一步？

你手上有 GitHub 上 star 过千的多变量时间序列异常检测模型（比如 USAD、MTAD-GAT、TranAD），本地环境配好了，PyTorch 版本对得上，连pip install -r requirements.txt都没报错——但一运行python train.py --dataset swat，就卡在FileNotFoundError: [Errno 2] No such file or directory: 'data/swat/train.npy'。不是模型写得烂，是数据集根本没按它要的格式准备好。这个标题里的「整理」二字，不是指把几个.csv文件拖进文件夹就完事；而是要把原始采集日志、工业传感器原始报文、IoT 设备导出的乱序时间戳数据，统一成带明确时间索引、对齐采样率、标注好异常区间、适配 PyTorch DataLoader 的np.ndarray或torch.Tensor结构。它解决的是：复现论文结果的第一道硬门槛、工业现场部署前的数据预处理标准化问题、以及跨数据集做消融实验时的 baseline 对齐需求。适合正在调试 SOTA 模型却反复栽在ValueError: expected 3D input (batch, channels, length)的算法工程师，也适合需要把产线 PLC 数据喂进检测模型但被 CSV 时间列格式搞崩溃的现场工程师。

2. 从原始数据到可训练张量：四步标准化流水线

多变量时间序列异常检测的数据整理，核心矛盾在于「原始数据形态极度碎片化」与「模型输入接口极度刚性」之间的冲突。常见原始数据包括：SCADA 系统导出的 Excel 表（含不规则空行、合并单元格、中文列名）、Modbus TCP 抓包的二进制日志、Prometheus 导出的__name__="cpu_usage"格式文本、甚至某次实验手写的 CSV（时间列是"2023-05-12 14:23:01.892"，而另一份是"1683897781892"毫秒时间戳）。我们不能靠人工改，必须建一套可复现、可版本化、可嵌入 CI/CD 的整理流水线。下面以最典型的 SWaT 数据集（水处理系统）和自定义工业传感器数据为例，拆解四步必经流程。

2.1 原始数据清洗：用 Pandas 处理“脏时间戳”与缺失值

SWaT 官方提供的SWaT_Dataset_Normal_v1.csv和SWaT_Dataset_Attack_v0.csv是典型「伪结构化」数据：首行是英文列名，但第 2 行开始混有非数值字符（如Normal、Attack字样），时间列Date和Time是分离的字符串，且存在大量NaN和?占位符。直接pd.read_csv()会把整列读成object类型，后续无法做数值计算。

import pandas as pd import numpy as np def clean_swat_raw(csv_path: str) -> pd.DataFrame: # 步骤1：跳过非数据行，指定列名（官方文档未提供完整列名，需手动确认） col_names = [ "Date", "Time", "FIT101", "LIT101", "MV101", "P101", "P102", "AIT201", "AIT202", "AIT203", "FIT201", "DPIT201", "PV201", "AIT301", "AIT302", "AIT303", "FIT301", "LIT301", "MV301", "MV302", "MV303", "MV304", "P301", "P302", "AIT401", "AIT402", "FIT401", "LIT401", "P401", "P402", "P403", "P404", "UV401", "AIT501", "AIT502", "AIT503", "AIT504", "FIT501", "FIT502", "FIT503", "FIT504", "P501", "P502", "P503", "AIT601", "AIT602" ] # 步骤2：读取时跳过前3行（含标题行和说明行），强制指定列数防错 df = pd.read_csv( csv_path, skiprows=3, # 跳过前3行非数据行 names=col_names, na_values=["?", "nan", "", "NaN"], # 显式声明缺失值标识 low_memory=False ) # 步骤3：拼接 Date+Time 为标准 datetime，并设为索引 df["datetime"] = pd.to_datetime(df["Date"] + " " + df["Time"]) df = df.set_index("datetime").sort_index() # 步骤4：数值列强制转 float，失败则置 NaN（如某些行含 'Normal' 字符） numeric_cols = df.columns.drop(["Date", "Time"]) for col in numeric_cols: df[col] = pd.to_numeric(df[col], errors="coerce") return df # 执行清洗 raw_df = clean_swat_raw("raw_data/SWaT_Dataset_Normal_v1.csv") print(f"清洗后形状: {raw_df.shape}, 时间范围: {raw_df.index[0]} ~ {raw_df.index[-1]}")

逻辑说明：skiprows=3是关键，SWaT 原始 CSV 前3行分别是标题、单位行、分隔线，不跳过会导致列名错位；na_values显式覆盖所有可能的缺失值符号，避免?被当字符串保留；pd.to_numeric(..., errors="coerce")是血泪经验——某次因FIT101列混入'Normal'字符串，导致整列变object，后续df.fillna(method="ffill")直接报错，加coerce后自动转NaN可继续处理。

2.2 时间对齐与重采样：解决多源传感器采样率不一致问题

工业现场常见场景：温度传感器每 5 秒上报一次，压力变送器每 200ms 发一帧，PLC 控制周期是 10ms。若直接拼接，时间索引长度不等、时间点不重合，pd.concat()会生成大量NaN。必须统一到一个基准采样率（如 1s），并选择插值策略。

def align_and_resample( df: pd.DataFrame, target_freq: str = "1S", method: str = "linear" ) -> pd.DataFrame: """ 对齐时间索引并重采样 :param target_freq: 目标频率，如 "1S"(1秒), "100L"(100毫秒) :param method: 插值方法，"linear"线性, "nearest"最近邻, "pad"前向填充 """ # 步骤1：确保索引是 DatetimeIndex if not isinstance(df.index, pd.DatetimeIndex): raise ValueError("Index must be DatetimeIndex") # 步骤2：先用 pad 填充短时缺失（如单个点丢失），再重采样 df_filled = df.fillna(method="pad") # 防止重采样时引入过多 NaN # 步骤3：重采样 —— 注意：resample() 后需聚合，此处用 first() 取首个值（因已填充） # 若原始数据是高频，也可用 mean() 降噪 df_resampled = df_filled.resample(target_freq).first() # 步骤4：对仍存在的 NaN，用指定方法插值（linear 最常用） df_interpolated = df_resampled.interpolate(method=method, limit_direction="both") return df_interpolated # 应用重采样 aligned_df = align_and_resample(raw_df, target_freq="1S", method="linear") print(f"重采样后形状: {aligned_df.shape}, 采样率: {pd.infer_freq(aligned_df.index)}")

参数说明：target_freq="1S"是工业场景保守选择——太密（如"100L"）会放大噪声，太疏（如"5S"）可能漏掉瞬态异常；method="linear"适合平缓变化的物理量（温度、液位），对开关量（阀门开/关）应改用"nearest"；limit_direction="both"是关键，否则默认只向前插值，末尾大段NaN无法修复。

2.3 异常标注标准化：把“攻击描述文本”转成 0/1 标签序列

SWaT 的攻击数据集（SWaT_Dataset_Attack_v0.csv）只在文件末尾附了一段文字说明：“Attack 1: 2015-12-29 10:00:00 to 2015-12-29 10:59:59, Manipulation of MV101...”。但模型需要每个时间步一个is_anomaly标签（0 正常 / 1 异常）。必须解析这段文本，生成与主数据表对齐的布尔序列。

import re from datetime import datetime def parse_attack_intervals(attack_desc: str) -> list: """ 解析攻击描述文本，返回 [(start_dt, end_dt, attack_type), ...] 示例输入: "Attack 1: 2015-12-29 10:00:00 to 2015-12-29 10:59:59, Manipulation..." """ intervals = [] # 匹配 "Attack X: YYYY-MM-DD HH:MM:SS to YYYY-MM-DD HH:MM:SS" pattern = r"Attack \d+: (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) to (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})" matches = re.findall(pattern, attack_desc) for start_str, end_str in matches: start_dt = datetime.strptime(start_str, "%Y-%m-%d %H:%M:%S") end_dt = datetime.strptime(end_str, "%Y-%m-%d %H:%M:%S") intervals.append((start_dt, end_dt)) return intervals def generate_anomaly_labels( df: pd.DataFrame, attack_intervals: list ) -> np.ndarray: """ 为 DataFrame 生成 anomaly label array (0/1) :param df: 已清洗、对齐的时间序列 DataFrame :param attack_intervals: parse_attack_intervals 返回的列表 :return: shape=(len(df),) 的 int64 数组 """ labels = np.zeros(len(df), dtype=np.int64) time_index = df.index for start_dt, end_dt in attack_intervals: # 找到时间索引中落在 [start_dt, end_dt] 内的位置 mask = (time_index >= start_dt) & (time_index <= end_dt) labels[mask] = 1 return labels # 实际使用（需先读取攻击描述文本） with open("raw_data/attack_description.txt", "r") as f: desc_text = f.read() intervals = parse_attack_intervals(desc_text) anomaly_labels = generate_anomaly_labels(aligned_df, intervals) print(f"异常点比例: {anomaly_labels.mean():.4f} ({anomaly_labels.sum()}/{len(anomaly_labels)})")

逻辑说明：parse_attack_intervals()用正则而非split(":")，因为攻击描述中冒号可能出现在“Manipulation of MV101: valve stuck”里，硬切会错位；generate_anomaly_labels()中mask计算依赖DatetimeIndex的向量化比较，比循环for i, t in enumerate(time_index)快 100 倍以上；标签数组anomaly_labels后续将与特征矩阵X一起存为.npy，供 DataLoader 加载。

2.4 保存为模型友好格式：.npy+.pkl元信息双保险

绝大多数 PyTorch 模型代码（如 USAD 的data_loader.py）期望数据是train.npy（shape(N, T, V)，N 样本数、T 时间步长、V 变量数）和test_label.npy（shape(N, T)）。但直接np.save("train.npy", X_train)会丢失关键元信息：原始变量名、采样率、时间范围。一旦后续要 debug 或复现实验，光看.npy文件等于黑匣子。因此必须配套保存.pkl元数据。

import pickle import numpy as np def save_dataset( X_train: np.ndarray, X_test: np.ndarray, y_test: np.ndarray, meta_info: dict, output_dir: str = "data/swat" ): """ 保存整理后的数据集 :param X_train: 训练特征 (N, T, V) :param X_test: 测试特征 (N, T, V) :param y_test: 测试标签 (N, T) 或 (N*T,) :param meta_info: 元信息字典，含 keys: ["var_names", "freq", "time_range", "description"] """ import os os.makedirs(output_dir, exist_ok=True) # 保存为 .npy（模型直接加载） np.save(f"{output_dir}/train.npy", X_train) np.save(f"{output_dir}/test.npy", X_test) np.save(f"{output_dir}/test_label.npy", y_test) # 保存元信息为 .pkl（人类可读+程序可解析） with open(f"{output_dir}/meta.pkl", "wb") as f: pickle.dump(meta_info, f) print(f"✅ 数据集已保存至 {output_dir}/") print(f" train.npy: {X_train.shape}, test.npy: {X_test.shape}, test_label.npy: {y_test.shape}") print(f" 元信息已存入 meta.pkl，可用 pickle.load() 读取") # 构建元信息（示例） meta = { "var_names": list(aligned_df.columns), "freq": "1S", "time_range": (aligned_df.index[0], aligned_df.index[-1]), "description": "SWaT dataset v1, cleaned and resampled to 1-second frequency", "anomaly_ratio": float(anomaly_labels.mean()) } # 假设已完成滑动窗口切分（见 3.1 节） # X_train, X_test, y_test = ... # save_dataset(X_train, X_test, y_test, meta, "data/swat")

参数说明：meta_info["var_names"]必须是list类型（非Index），否则pickle保存后load()可能报错；"freq"存字符串"1S"而非pd.Timedelta，保证跨 Python 版本兼容；"time_range"存datetime元组，比存字符串更易做时间计算；.pkl文件虽小（通常 <1KB），但它是后续排查“为什么模型在 SWaT 上 AUC 低而在 WADI 上高”的唯一线索——可能只是 WADI 的freq实际是"5S"，而代码里硬编码了"1S"。

3. 滑动窗口切分与归一化：让数据真正适配模型输入

整理好的aligned_df是一个(T_total, V)的二维表，但模型（如 LSTM、Transformer）需要三维输入(N, T_window, V)，其中N是样本数，T_window是时间窗口长度（如 100 步）。此外，不同变量量纲差异巨大（压力 0~10MPa，电流 0~20mA，温度 0~100℃），不归一化会导致梯度爆炸或收敛缓慢。这两步必须严格按顺序执行：先切窗口，再归一化，否则用全局 min/max 归一化后再切窗，会泄露未来信息（leakage）。

3.1 滑动窗口构造：用numpy.lib.stride_tricks.sliding_window_view避免内存爆炸

传统做法是for i in range(len(df)-window_size): X[i] = df.iloc[i:i+window_size].values，当T_total=1e6、V=50、window_size=100时，X占用内存约1e6 * 100 * 50 * 8 / 1e9 ≈ 40GB，远超常规机器内存。sliding_window_view是 NumPy 1.20+ 提供的零拷贝视图，内存占用恒为O(T_total * V)。

from numpy.lib.stride_tricks import sliding_window_view def create_sliding_windows( df: pd.DataFrame, window_size: int = 100, step: int = 1 ) -> np.ndarray: """ 创建滑动窗口视图，返回 shape (N, T, V) 的 ndarray :param df: 输入 DataFrame，index 为 DatetimeIndex :param window_size: 窗口长度（时间步数） :param step: 步长（默认1，即无重叠；step=10 则每10步取一个窗） :return: shape (N, window_size, V) 的 float32 数组 """ # 提取数值矩阵，确保 dtype 为 float32 节省内存 data_matrix = df.values.astype(np.float32) # (T_total, V) # 使用 sliding_window_view 创建 (T_total, window_size, V) 视图 # 注意：axis=0 是时间轴，所以 window 在 axis=0 上滑动 windows = sliding_window_view(data_matrix, window_shape=window_size, axis=0) # windows.shape = (T_total - window_size + 1, window_size, V) # 按 step 采样，减少样本数（可选） if step > 1: windows = windows[::step] return windows # 示例：为 SWaT 构建 100 步窗口 X_all = create_sliding_windows(aligned_df, window_size=100, step=10) print(f"滑动窗口后形状: {X_all.shape} (样本数, 时间步, 变量数)")

逻辑说明：sliding_window_view(..., axis=0)是关键，若误写axis=1（变量轴），会得到(T_total, V-window_size+1, window_size)，完全错误；step=10不是为了提速，而是防止过拟合——连续窗口间重叠度过高（99%），模型学到的是“相邻窗口相似”，而非“时间模式”；返回float32而非float64，因深度学习模型权重默认float32，双精度徒增内存无益。

3.2 分割训练/测试集：按时间顺序切分，禁用随机打乱

时间序列的核心约束是未来不可知。若用sklearn.model_selection.train_test_split(X, test_size=0.2, shuffle=True)，会把未来的数据混入训练集，导致模型在测试集上表现虚高（data leakage）。必须严格按时间顺序：前 80% 时间点为训练，后 20% 为测试。

def split_by_time( X: np.ndarray, y: np.ndarray = None, train_ratio: float = 0.8 ) -> tuple: """ 按时间顺序分割，返回 (X_train, X_test, y_train, y_test) :param X: shape (N, T, V) 的窗口数组 :param y: shape (N, T) 或 (N,) 的标签数组（可选） :param train_ratio: 训练集占比 :return: 四元组 """ n_total = len(X) n_train = int(n_total * train_ratio) X_train, X_test = X[:n_train], X[n_train:] if y is not None: if y.ndim == 2: # (N, T) 标签 y_train, y_test = y[:n_train], y[n_train:] else: # (N,) 标签，如每个窗口一个整体标签 y_train, y_test = y[:n_train], y[n_train:] return X_train, X_test, y_train, y_test else: return X_train, X_test, None, None # 应用分割 X_train, X_test, _, y_test_windows = split_by_time( X_all, y=anomaly_labels, # 注意：y 是 (T_total,)，需先映射到窗口级 train_ratio=0.8 ) # 将 (T_total,) 标签映射到 (N, T) 窗口标签：每个窗口内只要有一个异常点，整个窗口标为异常 def labels_to_window_labels( full_labels: np.ndarray, # (T_total,) window_size: int, step: int = 1 ) -> np.ndarray: # (N, T) n_windows = (len(full_labels) - window_size) // step + 1 window_labels = np.zeros((n_windows, window_size), dtype=np.int64) for i in range(n_windows): start_idx = i * step end_idx = start_idx + window_size window_labels[i] = full_labels[start_idx:end_idx] return window_labels y_test_windows = labels_to_window_labels(anomaly_labels, window_size=100, step=10) print(f"X_train: {X_train.shape}, X_test: {X_test.shape}, y_test_windows: {y_test_windows.shape}")

参数说明：train_ratio=0.8是经验阈值，低于 0.7 训练数据不足，高于 0.9 测试集太小难评估；labels_to_window_labels()中window_labels[i] = full_labels[start_idx:end_idx]是逐窗口复制，而非any()判断——因多数模型（如 USAD）需要逐时间步的异常概率，而非窗口级二分类。

3.3 归一化策略选择：为什么 MinMaxScaler 在异常检测中可能失效？

常规做法是from sklearn.preprocessing import MinMaxScaler; scaler.fit_transform(X_train.reshape(-1, V))，但异常检测场景下，异常点本身会扭曲 scaler 的min_和max_。例如某压力传感器正常范围 0~10MPa，但攻击时被注入 100MPa 噪声，MinMaxScaler的max_变成 100，导致正常数据被压缩到[0, 0.1]，模型难以分辨细微漂移。更鲁棒的做法是：

• 方案A（推荐）：用训练集的 IQR（四分位距）截断后归一化
  先剔除训练集中Q1-1.5*IQR到Q3+1.5*IQR之外的离群点（假设它们是未标注的异常），再用剩余数据的 min/max 归一化。

• 方案B：RobustScaler（中位数+IQR）
  不受极值影响，但输出范围不固定（非[0,1]），需后续调整。

from sklearn.preprocessing import RobustScaler import numpy as np def robust_normalize( X_train: np.ndarray, X_test: np.ndarray, quantile_range: tuple = (25, 75) ) -> tuple: """ 使用 RobustScaler 归一化，抗异常点干扰 :param X_train: (N, T, V) :param X_test: (N, T, V) :param quantile_range: 四分位数范围，默认 (25, 75) :return: 归一化后的 (X_train_norm, X_test_norm) """ n_train, T, V = X_train.shape n_test = X_test.shape[0] # 展平为 (N*T, V) 以便 scaler 处理 X_train_flat = X_train.reshape(-1, V) # (N*T, V) X_test_flat = X_test.reshape(-1, V) # (N*T, V) # 初始化 RobustScaler scaler = RobustScaler(quantile_range=quantile_range) # 仅用训练集拟合（绝不 fit 测试集！） X_train_flat_scaled = scaler.fit_transform(X_train_flat) X_test_flat_scaled = scaler.transform(X_test_flat) # 恢复三维形状 X_train_norm = X_train_flat_scaled.reshape(n_train, T, V) X_test_norm = X_test_flat_scaled.reshape(n_test, T, V) return X_train_norm, X_test_norm # 执行归一化 X_train_norm, X_test_norm = robust_normalize(X_train, X_test) print(f"归一化后 X_train_norm.min(): {X_train_norm.min():.4f}, max(): {X_train_norm.max():.4f}")

避坑提示：scaler.fit_transform()必须只在X_train_flat上调用，X_test_flat只能transform()；reshape(-1, V)是为了按变量维度归一化（每个变量独立缩放），而非按时间步；quantile_range=(25,75)是默认值，若数据偏态严重（如大量 0 值），可试(10,90)扩大范围。

4. 多变量数据集避坑指南：5 条血泪换来的经验

整理多变量时间序列数据集不是体力活，而是处处埋雷的精细活。以下 5 条是我在模拟项目 X（某高校实验室水处理平台）和某跨平台系统（工业 IoT 边缘检测框架）中踩过的坑，每一条都曾导致模型 AUC 下降 15% 以上，或训练直接中断。

4.1 现象：ValueError: Found array with dim 3. Expected dim <= 2.

原因：在调用sklearn.preprocessing.StandardScaler时，误将三维数组X_train（shape(N,T,V)）直接传入，而StandardScaler只接受二维输入（fit(X)要求X.shape=(n_samples, n_features)）。
解决：必须先reshape成二维，归一化后再reshape回三维。正确写法：

X_train_flat = X_train.reshape(-1, X_train.shape[-1]) # (N*T, V) scaler = StandardScaler().fit(X_train_flat) X_train_norm = scaler.transform(X_train_flat).reshape(X_train.shape)

4.2 现象：模型在训练集上 loss 持续下降，但在测试集上 AUC 为 0.5（随机猜测）

原因：时间序列分割时用了shuffle=True，导致训练集混入未来数据，模型学到了“时间穿越”能力，但实际部署时失效。
解决：删除所有shuffle参数，严格按X_train, X_test = X[:n_train], X[n_train:]切分；并在数据加载器中添加断言：

assert np.all(X_test[:, 0, 0] > X_train[-1, -1, 0]), "Time leakage detected!"

4.3 现象：train.npy加载后np.isnan(X_train).sum() > 0，但训练时报RuntimeError: Input tensor contains NaN values

原因：清洗阶段用df.fillna(method="ffill")填充了开头的NaN，但若第一行就是NaN，ffill无效，该列开头仍为NaN；重采样时resample().first()遇到全NaN区间会返回NaN。
解决：在清洗最后一步强制处理开头NaN：

df = df.fillna(method="ffill").fillna(method="bfill") # bfill 处理开头 # 或更安全：用插值填补所有 NaN df = df.interpolate(method="linear", limit_direction="both")

4.4 现象：anomaly_labels全为 0，但原始攻击描述明确写了时间区间

原因：攻击描述中的时间字符串格式与aligned_df.index的datetime格式不匹配。例如描述是"2015-12-29 10:00:00"，而aligned_df.index[0]是Timestamp('2015-12-29 10:00:00.000000000')，>=比较时因纳秒精度被截断，导致mask全False。
解决：统一时间精度，在生成mask前将start_dt/end_dt转为与索引相同精度：

start_ts = pd.Timestamp(start_dt).floor("S") # 向下取整到秒 end_ts = pd.Timestamp(end_dt).ceil("S") # 向上取整到秒 mask = (time_index >= start_ts) & (time_index <= end_ts)

4.5 现象：sliding_window_view报ValueError: window_shape cannot be larger than the size of the axis

原因：window_size=100，但aligned_df经过重采样后只剩 50 行（如原始数据只有 50 秒），sliding_window_view要求window_size <= len(axis)。
解决：在调用前加长度检查，并给出明确报错：

if len(df) < window_size: raise ValueError( f"Data length {len(df)} < window_size {window_size}. " f"Please check if resampling dropped too many rows." )

5. 进阶技巧：用Dask处理超大规模数据集（TB 级）

当你的数据来自一年的风电机组 SCADA 日志（每台风机 200+ 传感器，10ms 采样），单机内存无法容纳aligned_df时，pandas会直接 OOM。此时必须切换到Dask——一个能将 DataFrame 拆分成块（partitions）、惰性计算、并行处理的库。核心思路是：不把全部数据 load 进内存，而是在磁盘上分块处理，最后合并结果。

5.1 Dask DataFrame 替代 Pandas：分块读取与清洗

import dask.dataframe as dd def clean_large_dataset_dask( csv_path: str, chunksize: int = 100000 ) -> dd.DataFrame: """ 用 Dask 分块读取并清洗超大 CSV :param csv_path: 原始 CSV 路径 :param chunksize: 每块行数，根据内存调整（建议 5e4~5e5） :return: Dask DataFrame """ # 步骤1：用 dd.read_csv 读取，指定 dtypes 避免类型推断开销 dtypes = {col: "float64" for col in [ "FIT101", "LIT101", "MV101", "P101", "P102", "AIT201", "AIT202" ]} dtypes.update({"Date": "string", "Time": "string"}) df = dd.read_csv( csv_path, blocksize=f"{chunksize}MB", # 按文件大小分块，非行数 dtype=dtypes, assume_missing=True ) # 步骤2：Dask 版清洗（惰性，不立即执行） df["datetime"] = dd.to_datetime(df["Date"] + " " + df["Time"]) df = df.set_index("datetime").sort_index() # 步骤3：数值列转 float（Dask 不支持 errors="coerce"，改用 map_partitions） def safe_to_numeric(partition): for col in partition.select_dtypes(include=["object"]).columns: partition[col] = dd.to_numeric(partition[col], errors="coerce") return partition df = df.map_partitions(safe_to_numeric) return df # 执行（不触发计算） large_df = clean_large_dataset_dask("raw_data/wind_turbine_full_year.csv") print(f"Dask DataFrame partitions: {large_df.npartitions}")

关键差异：dd.read_csv()返回的是计算图，print(large_df)只显示 schema；df.map_partitions()是对每个分区应用函数，比df.apply()更高效；blocksize="100MB"比sample_nrows更稳定，避免因行长短不一导致分块不均。

5.2 Dask 版时间对齐与重采样：避免.compute()全量加载

Dask的resample()不支持DatetimeIndex直接操作，需先用set_index()确保索引是datetime，再用map_partitions在每个分区内部重采样，最后repartition()对齐时间。

def dask_resample_to_freq( df: dd.DataFrame, freq: str = "1S" ) -> dd.DataFrame: """ Dask 版重采样 <p> <a href="https://download.csdn.net/download/banana/92319972" style="color:#ec7500;font-size:14px;"> 本文还有配套的精品资源，点击获取 </a> <img alt="menu-r.4af5f7ec.gif" src="https://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif" style="width:16px;margin-left:4px;vertical-align:text-bottom;cursor:text;"> </p>