模块 basalt.data¶

6 个公开类型

BasaltD 数据管线：流式与内存映射两条读取路径、标准化参数。

类型概览

final class DataLoader¶

BasaltD 记录的批式读取入口，附带统计与写盘工具。

文件头固定 16 字节：magic 0x4A534C44（"BasaltD"）、version 1、featureCount 与保留位（写入为 0，读取要求为 0）。其后每条记录为 featureCount 个 double 特征加一个 double 标签，记录宽度 (featureCount + 1) * 8 字节。

两个读取器共享同一 Batch 契约。BatchStream 顺序读入按批大小分配的直接缓冲区，再用 DoubleBuffer 视图解码；MappedBatchStream 映射整个文件，以绝对批量读取汇集行，映射之后一个 epoch 内无系统调用。批次、特征数组、标签数组与字节缓冲只在构造期分配一次。

除读取器外，类还提供三个静态工具：write 生成 BasaltD 文件，estimateNormalization 单遍统计均值与样本标准差，rowCount 由文件大小推算记录数。

示例

Path file = Path.of("dataset.bsld");
Matrix features = Matrix.fromRows(new double[][]{
        {1.0, 2.0}, {3.0, 4.0}, {5.0, 7.0}, {2.0, 1.0}});
DataLoader.write(file, features, new double[]{0.0, 1.0, 1.0, 0.0});

Normalization normalization = DataLoader.estimateNormalization(file, 2);
try (DataLoader.BatchStream stream = DataLoader.builder(file)
        .featureCount(2)
        .batchSize(2)
        .normalization(normalization)
        .open()) {
    while (stream.next()) {
        DataLoader.Batch batch = stream.batch();
        for (int row = 0; row < batch.rows(); row++) {
            double label = batch.label(row);
            double first = batch.feature(row, 0);
        }
    }
}

注意

类本身无实例（构造器私有），Builder、Batch、BatchStream、MappedBatchStream 均为静态嵌套类型。

注意

读取器只产出 double[]，不构造张量、不涉及计算图；需要矩阵时由调用方转换。

注意

批次数组在流的生命周期内复用，next() 会覆盖上一批内容，跨批保存须自行拷贝。

注意

每个流都不是线程安全的，并发读取应为每个线程建立独立的流。

参见：basalt.data.Normalization, basalt.data.DataLoader.Builder, basalt.data.DataLoader.Batch, basalt.data.DataLoader.BatchStream, basalt.data.DataLoader.MappedBatchStream, basalt.io.ModelCodec

方法

builder(Path source) → Builderstatic¶

为指定 BasaltD 文件创建读取器配置。

注意

此时不打开文件，路径合法性推迟到 open()/openMapped() 才校验

参数

• source (Path) – BasaltD 文件路径；为 null 时构造 Builder 立即抛 NullPointerException

返回

尚未设置 featureCount 的 Builder，后续用 open() 或 openMapped() 打开流

estimateNormalization(Path source, int featureCount) → Normalizationstatic¶

单遍统计各特征均值与样本标准差。

内部以 batchSize 固定为 1024 的顺序读取器遍历全部记录，用 Welford 增量公式累计均值与二阶中心矩，不缓存整份数据；标准差按样本口径计算，即除以 count - 1。

样本数少于 2 时抛 IllegalArgumentException。读取失败或文件头与 featureCount 不匹配时抛 UncheckedIOException。标准差为 0 的特征在后续应用时只做中心化。

注意

只读取特征列，标签列被跳过

注意

统计过程复用单个批次数组，内存占用与记录数无关

参数

• source (Path) – BasaltD 文件路径

• featureCount (int) – 每条记录的特征数，必须为正且与文件头中的值一致

返回

持有均值和标准差副本的不可变 Normalization

write(Path target, Matrix features, double[] labels)static¶

将特征矩阵与标签写成 BasaltD 文件。

以 CREATE、WRITE、TRUNCATE_EXISTING 打开目标文件，先写 16 字节头（magic、version 1、features.columns()、保留位 0），再逐行写出特征与标签。缓冲容量取 1 MiB 与两条记录宽度中的较大者，写满即刷出。

features 或 labels 为 null 时抛 NullPointerException；labels.length 与 features.rows() 不等时抛 IllegalArgumentException；通道打开或写入失败时抛 UncheckedIOException。记录宽度计算溢出 int 时抛 ArithmeticException。

注意

每条记录先在复用的一维数组内拼装再整体写入缓冲

注意

写入的浮点数为 ByteBuffer 默认的大端序

参数

• target (Path) – 输出路径，已存在的文件会被截断覆盖

• features (Matrix) – 特征矩阵，行数即记录数，列数即文件头中的 featureCount

• labels (double[]) – 标签数组，长度必须等于 features.rows()，逐行写入记录末尾

rowCount(Path source, int featureCount) → longstatic¶

按文件大小推算记录数。

取文件字节数减去 16 字节头，再除以 (featureCount + 1) * 8。余数非零或结果载荷为负时视为截断，抛 UncheckedIOException（cause 为 IOException）；文件不存在或不可读同样包装为 UncheckedIOException。

实现只依赖 Files.size，不打开通道、不读取也不校验 magic、version 与保留位，因此对头部损坏的文件仍可能返回一个数值。

注意

需要头部校验时改用 builder(source).featureCount(...).open()，其会比对文件头

注意

结果以 long 返回，但 openMapped() 只接受不超过 Integer.MAX_VALUE 的行数

参数

• source (Path) – BasaltD 文件路径

• featureCount (int) – 每条记录的特征数，用于推导记录宽度；加一后乘 8 溢出 int 时抛 ArithmeticException

返回

记录条数，未做任何上界限制

final class Normalization¶

不可变的标准化参数，应用时零分配。

实例保存每特征均值、标准差，以及构造时预先算好的标准差倒数。应用公式为 (x - mean) * inverseStd，原地写回传入数组，不产生临时对象。标准差为 0 时倒数取 1.0，该特征只做中心化，不会产生无穷大。

构造入口只有 standardize 与 DataLoader.estimateNormalization。standardize 会克隆两个入参数组，means() 与 standardDeviations() 也返回副本，因此外部对数组的修改不会影响实例，实例可安全共享。

示例

Path file = Path.of("dataset.bsld");
DataLoader.write(file,
        Matrix.fromRows(new double[][]{{1.0, 2.0}, {3.0, 4.0}, {5.0, 7.0}}),
        new double[]{0.0, 1.0, 0.0});

Normalization normalization = DataLoader.estimateNormalization(file, 2);
int features = normalization.featureCount();
double[] means = normalization.means();
double[] deviations = normalization.standardDeviations();

注意

apply 是包级方法，仅由 BatchStream 与 MappedBatchStream 在解码一行后调用，库外无法直接调用。

注意

实例不可变、无状态，可跨线程共享；但 apply 会就地覆盖调用方数组。

注意

特征数在构造时固定，不能对长度不同的记录复用同一实例。

参见：basalt.data.DataLoader.estimateNormalization(Path, int), basalt.data.DataLoader.Builder.normalization(Normalization)

方法

standardize(double[] means, double[] standardDeviations) → Normalizationstatic¶

由均值与标准差构造标准化参数。

两个数组长度必须相同且非空，所有元素必须有限，标准差还须非负，否则抛 IllegalArgumentException。通过校验后克隆入参并预计算倒数，返回的实例与入参数组不再共享存储。

标准差为 0 的位置倒数记为 1.0，使该特征在应用时减去均值后保持原值。

注意

负数、NaN 与无穷都会被拒绝，不会静默传播到训练数据

参数

• means (double[]) – 每个特征的均值，长度即特征数，不能为空

• standardDeviations (double[]) – 每个特征的标准差，长度须与 means 一致，取值须有限且非负

返回

持有两份副本与倒数表的不可变实例

featureCount() → int¶

特征数，等于均值数组长度。

注意

Builder 用该值校验归一化参数与读取器配置是否匹配

返回

构造时确定的特征数

means() → double[]¶

返回均值的防御性副本。

注意

每次调用都克隆，循环中反复调用会产生额外分配

返回

长度为 featureCount() 的新数组，修改不影响实例

standardDeviations() → double[]¶

返回标准差的防御性副本。

注意

返回的是标准差本身，不是内部使用的倒数

返回

长度为 featureCount() 的新数组，元素非负

static final class DataLoader.Batch¶

单批数据的可复用视图，随读取器推进被覆盖。

特征以行优先方式存放在长度为 featureCount * batchSize 的一维数组中，标签数组长度为 batchSize。两个数组在构造时按批大小一次性分配，之后每次 next() 只改写前部内容，rows() 指明本批真正有效的行数。

有效区间为：特征前 rows() * featureCount() 个元素，标签前 rows() 个元素。数组尾部残留上一批的数据，不应读取。实例由读取器创建并长期持有，调用方只做只读访问。

示例

try (DataLoader.BatchStream stream = DataLoader.builder(Path.of("dataset.bsld"))
        .featureCount(2)
        .batchSize(64)
        .open()) {
    while (stream.next()) {
        DataLoader.Batch batch = stream.batch();
        for (int row = 0; row < batch.rows(); row++) {
            double label = batch.label(row);
            double first = batch.feature(row, 0);
        }
    }
}

注意

features() 与 labels() 返回内部数组本身，不复制，跨批保留引用会读到被覆盖的内容。

注意

实例与流一一绑定，非线程安全；批次内容在下次 next() 后失效。

注意

需要跨批留存时，用 System.arraycopy 复制有效区间。

参见：basalt.data.DataLoader.BatchStream.batch(), basalt.data.DataLoader.MappedBatchStream.batch()

方法

rows() → int¶

当前批的有效行数。

注意

末批通常小于 batchSize，遍历必须以该值为上界

返回

本批行数，取值不超过 batchSize；流耗尽后为 0

featureCount() → int¶

每条记录的特征数。

返回

构造流时指定的特征数，整条流内不变

features() → double[]¶

返回内部特征数组，行优先且复用。

数组长度恒为 featureCount() * batchSize，仅前 rows() * featureCount() 个元素属于当前批。已配置归一化时，数组中是标准化后的值，而非文件原始值。

注意

下一次 next() 会覆盖前部内容

注意

直接传给需要整块特征的 API 时，须按 rows() * featureCount() 截取

返回

内部数组本身，不复制

labels() → double[]¶

返回内部标签数组，复用。

数组长度恒为 batchSize，仅前 rows() 个元素属于当前批；标签不参与归一化，与文件中的值一致。

注意

下一次 next() 会覆盖前部内容

返回

内部数组本身，不复制

feature(int row, int feature) → double¶

读取批内某行某列的特征值。

以 row * featureCount() + feature 定位数组元素；越界检查同时约束行与列，row 不小于 rows() 或 feature 不小于 featureCount() 时抛 IndexOutOfBoundsException。

注意

逐元素访问有下标运算开销，整批处理时优先取 features() 数组

参数

• row (int) – 批内行号，有效范围 [0, rows())

• feature (int) – 特征下标，有效范围 [0, featureCount())

返回

该位置的特征值，已应用归一化（若配置）

label(int row) → double¶

读取批内某行的标签。

row 超出 [0, rows()) 时抛 IndexOutOfBoundsException，错误信息与特征越界区分。

参数

• row (int) – 批内行号，有效范围 [0, rows())

返回

该行记录的标签值，未做归一化

static final class DataLoader.BatchStream implements AutoCloseable¶

顺序分批读取 BasaltD，可回绕重读。

构造时按 recordBytes * batchSize 分配直接缓冲区，并创建批次、特征数组与标签数组，之后不再分配。每次 next() 先压缩缓冲区，从通道循环读取直到缓冲填满或文件结束，再把完整的记录解码进批次；不足一条记录的尾部字节在文件结束时抛 EOFException。

rewind() 把通道位置重置到 16 字节头之后、清空缓冲区并复位结束标志，缓冲与批次对象继续复用，不重新打开文件。流在关闭后不可再用。

示例

try (DataLoader.BatchStream stream = DataLoader.builder(Path.of("dataset.bsld"))
        .featureCount(8)
        .batchSize(128)
        .open()) {
    for (int epoch = 0; epoch < 3; epoch++) {
        while (stream.next()) {
            DataLoader.Batch batch = stream.batch();
            int rows = batch.rows();
        }
        stream.rewind();
    }
}

注意

必须用 try-with-resources 或显式 close() 释放文件通道。

注意

shuffle 与 seed 只影响 openMapped()，本读取器始终按文件顺序返回。

注意

非线程安全，且不支持与另一读取器共享同一批次。

参见：basalt.data.DataLoader.Builder.open(), basalt.data.DataLoader.MappedBatchStream

方法

next() → boolean¶

解码下一批，流耗尽返回 false。

从通道补齐缓冲后按记录宽度切分，最多解码 batchSize 行，逐行写入特征数组、标签数组，并在配置了归一化时就地标准化该行。返回 false 时批次行数被置 0。

流已关闭时抛 IOException；文件尾剩余字节不是记录宽度的整数倍时抛 EOFException，用于识别截断文件。

注意

每次调用都复用同一 Batch 实例，返回的引用在下次调用后内容改变

返回

是否读到至少一行

抛出

• IOException – 流已关闭，或文件尾部存在不完整记录（EOFException）

batch() → Batch¶

返回与流绑定的批次视图。

注意

首次 next() 之前 rows() 为 0

返回

构造流时创建并持续复用的 Batch，永不为 null

rewind()¶

回到第一条记录，缓冲保持不变。

将通道位置设为 16，清空缓冲并复位结束标志，批次行数归零。缓冲区、批次与数组均保留，因此回绕不产生分配，也不重新解析文件头。

注意

回绕后仍需重新调用 next() 才能取到第一批

注意

归一化参数保持不变

抛出

• IOException – 流已关闭，或通道定位失败

close()¶

关闭底层文件通道。

首次调用置关闭标志并关闭通道，重复调用不做任何事。关闭后 next() 与 rewind() 均抛 IOException，批次数组仍可读取但内容为最后一次解码结果。

抛出

• IOException – 关闭通道时发生 I/O 错误

static final class DataLoader.Builder¶

两个读取器共享的配置构建器。

字段初值：featureCount 为 0（没有默认值，必须显式设置），batchSize 为 1024，normalization 为 null，shuffle 为 false，seed 为 42。所有设置方法返回 this，可链式调用。

open() 与 openMapped() 先校验配置：featureCount 或 batchSize 非正、归一化特征数不匹配时抛 IllegalArgumentException；随后打开通道并比对 16 字节头，I/O 失败一律包装为 UncheckedIOException。

示例

DataLoader.Builder builder = DataLoader.builder(Path.of("dataset.bsld"))
        .featureCount(8)
        .batchSize(256)
        .shuffle(true)
        .seed(7);
try (DataLoader.MappedBatchStream stream = builder.openMapped()) {
    while (stream.next()) {
        int rows = stream.batch().rows();
    }
}

注意

Builder 本身不持有打开的通道，可在多次 open() 之间复用，每次得到独立的流。

注意

非线程安全，构建完成后不应再修改。

注意

shuffle 与 seed 只对 openMapped() 生效，open() 会静默忽略。

参见：basalt.data.DataLoader.builder(Path), basalt.data.DataLoader.BatchStream, basalt.data.DataLoader.MappedBatchStream

方法

featureCount(int featureCount) → Builder¶

设置每条记录的特征数。

注意

未设置时 open()/openMapped() 抛 IllegalArgumentException

注意

与文件头不一致时读取阶段抛 UncheckedIOException（Invalid or incompatible BasaltD header）

参数

• featureCount (int) – 每条记录的特征数，无默认值（初值 0），必须为正且与文件头一致

返回

this，便于链式调用

batchSize(int batchSize) → Builder¶

设置每批最多行数，默认 1024。

注意

同时决定顺序读取器的直接缓冲容量与批次特征数组长度，过大将按 recordBytes * batchSize 一次性分配

参数

• batchSize (int) – 每批行数上限，默认 1024，必须为正；末批可能更少

返回

this，便于链式调用

normalization(Normalization normalization) → Builder¶

设置读取时应用的标准化参数。

注意

其 featureCount() 必须等于 featureCount 设置，否则 open() 抛 IllegalArgumentException

注意

只影响特征列，标签原样返回

参数

• normalization (Normalization) – 逐行就地应用的标准化参数；null 表示不归一化，也是默认值

返回

this，便于链式调用

shuffle(boolean shuffle) → Builder¶

是否按 epoch 打乱行序，默认 false。

注意

仅对 openMapped() 生效，open() 忽略该设置

注意

打乱只置换 int[] 索引数组，不移动映射区中的数据

参数

• shuffle (boolean) – true 时每个 epoch 置换行索引；默认 false 即按文件顺序

返回

this，便于链式调用

seed(long seed) → Builder¶

设置打乱所用随机种子，默认 42。

注意

仅对 openMapped() 且 shuffle 为 true 时有意义，相同 seed 得到相同的行序

参数

• seed (long) – 初始 epoch 的置换种子，默认 42；后续 epoch 使用 seed + epoch * 1000003

返回

this，便于链式调用

open() → BatchStream¶

打开顺序读取器。

以只读方式打开通道，读取并校验 16 字节头（magic、version、featureCount、保留位），然后按 recordBytes * batchSize 分配直接缓冲区与批次，返回 BatchStream。头部校验失败时通道被关闭并抛 UncheckedIOException；配置非法时在打开通道前抛 IllegalArgumentException。

注意

头部截断会以 EOFException 作为 cause 包装为 UncheckedIOException

注意

不检查文件长度是否为记录宽度的整数倍，截断要到读取尾部才暴露

返回

顺序读取器，需由调用方关闭

openMapped() → MappedBatchStream¶

打开内存映射读取器。

校验头部后把整个文件映射为只读 MappedByteBuffer，跳过 16 字节头得到 DoubleBuffer 切片，并由 (size - 16) / recordBytes 推出总行数。载荷长度不是记录宽度整数倍、行数超过 Integer.MAX_VALUE、或映射失败时抛 UncheckedIOException。

与 open() 的差异：数据由操作系统按页装入，不占用用户态缓冲，也不经过通道读取；批次与索引数组仍然分配，但之后每个 epoch 不再产生系统调用。

注意

映射区在流关闭后由 GC 回收，close() 不会立即解除映射

注意

文件在流使用期间被外部修改会导致内容与校验不一致，加载阶段不做检测

返回

内存映射读取器，需由调用方关闭

static final class DataLoader.MappedBatchStream implements AutoCloseable¶

内存映射读取器，按 epoch 打乱且不重复分配。

构造时把整个文件映射为只读缓冲，并跳过 16 字节头取出 DoubleBuffer 视图；同时建立长度为 totalRows 的 int 索引数组，初值为文件顺序，shuffle 开启时立刻做一次 Fisher-Yates 置换。

每次 next() 用绝对批量读取把记录宽度中的特征段复制进批次，标签单独取一个元素；游标按索引数组推进，一个 epoch 内不再访问通道，也没有字节缓冲。rewind() 让游标归零、epoch 自增，并按 seed + epoch * 1000003 重新置换索引。

示例

try (DataLoader.MappedBatchStream stream = DataLoader.builder(Path.of("dataset.bsld"))
        .featureCount(8)
        .batchSize(256)
        .shuffle(true)
        .seed(7)
        .openMapped()) {
    for (int epoch = 0; epoch < 5; epoch++) {
        while (stream.next()) {
            DataLoader.Batch batch = stream.batch();
            int rows = batch.rows();
        }
        stream.rewind();
    }
}

注意

批次、特征数组与标签数组各分配一次，之后只改写内容。

注意

打乱作用于索引数组，映射区中的原始顺序不变。

注意

close() 只置关闭标志，不显式解除映射；关闭后 next()/rewind() 抛 IllegalStateException（与 BatchStream 的 IOException 不同）。

注意

总行数受 int 索引限制，超过 Integer.MAX_VALUE 的文件在 openMapped() 阶段即被拒绝。

参见：basalt.data.DataLoader.Builder.openMapped(), basalt.data.DataLoader.BatchStream

方法

next() → boolean¶

汇集下一批，耗尽返回 false。

取 min(batchSize, totalRows - cursor) 行，按索引数组找到源记录，把特征段批量复制进批次数组并逐行取标签，配置了归一化时就地标准化。行数为 0 时返回 false 并把批次行数置 0。

本方法不声明受检异常；流已关闭时抛 IllegalStateException。

注意

不抛出 IOException：映射建立后读取只触及已映射内存

返回

是否读到至少一行

batch() → Batch¶

返回与流绑定的批次视图。

注意

内容在每次 next() 后被覆盖

返回

构造时创建并持续复用的 Batch，永不为 null

rewind()¶

开始新 epoch，必要时重新打乱。

游标归零、epoch 加一；shuffle 为 true 时用 seed + epoch * 1000003 新建 Random 并再次做 Fisher-Yates 置换，因此相邻 epoch 的顺序不同，而相同 epoch 号在相同 seed 下可复现。shuffle 为 false 时只重置游标。

不重新映射文件，也不重新读取文件头；流已关闭时抛 IllegalStateException。

注意

epoch 从 0 开始，首次 rewind() 后为 1

注意

置换就地复用同一 int 数组，不产生分配

totalRows() → int¶

映射到的总记录数。

注意

与 epoch 无关，整个生命周期内不变

返回

由 (文件大小 - 16) / 记录宽度 推出的行数

epoch() → int¶

已开始的 epoch 计数。

注意

该值同时参与下一轮置换种子的计算

返回

初始为 0，每次 rewind() 加一

close()¶

置关闭标志并停止后续读取。

只把内部标志设为已关闭，不关闭通道、不解除映射；映射区在缓冲区不可达后由 GC 回收。重复调用无副作用，关闭后再调用 next() 或 rewind() 抛 IllegalStateException。

注意

不声明受检异常，适合放在 try-with-resources 中使用