lamindb-data-management

LaminDB — 生物数据管理

概述

LaminDB 是一个面向生物学的开源数据框架，可让数据变得可查询、可追溯，并符合 FAIR（可发现、可访问、可互操作、可重用）原则。它结合了数据湖仓架构、谱系追踪、生物学本体验证，以及统一的 Python API，可管理从原始文件到已注释、已整理的数据制品的生物数据集。

适用场景

• 管理和版本控制生物数据集（scRNA-seq、空间数据、流式细胞术、多模态数据）
• 追踪计算谱系（哪些代码生成了哪些数据）
• 根据生物学本体（细胞类型、基因、组织、疾病）验证和整理数据
• 构建跨多个实验的可查询数据湖仓
• 通过自动捕获环境和来源信息来确保可复现性
• 与工作流管理器（Nextflow、Snakemake）或 MLOps 工具（W&B、MLflow）集成
• 使用基于本体的注释（Bionty）来标准化元数据
• 对于单细胞分析工作流（聚类、差异表达），请改用 scanpy
• 若仅需查询本体，而无需管理数据，请直接使用 bionty

前置条件

pip install lamindb
# With extras for specific data types
pip install 'lamindb[bionty,zarr,fcs]'

设置：使用前需要初始化实例：

lamin login
lamin init --storage ./my-data --name my-project
# Or with cloud storage:
# lamin init --storage s3://my-bucket --name my-project --db postgresql://...

实例类型：本地 SQLite（开发）、Cloud + SQLite（小型团队）、Cloud + PostgreSQL（生产环境）。

快速入门

import lamindb as ln

ln.track()  # Start lineage tracking

# Save an artifact
import pandas as pd
df = pd.DataFrame({"gene": ["TP53", "BRCA1"], "score": [0.95, 0.87]})
artifact = ln.Artifact.from_df(df, key="results/gene_scores.parquet", description="Gene importance scores")
artifact.save()
print(f"Saved: {artifact.uid}, size: {artifact.size}")

# Query artifacts
results = ln.Artifact.filter(key__startswith="results/").df()
print(f"Found {len(results)} artifacts")

ln.finish()

核心 API

1. Artifacts — 数据对象

Artifacts 是带有版本的数据对象（文件、DataFrame、AnnData、数组）。

import lamindb as ln
import pandas as pd
import anndata as ad

ln.track()

# From DataFrame
df = pd.DataFrame({"sample": ["A", "B"], "value": [1.5, 2.3]})
artifact = ln.Artifact.from_df(df, key="experiments/batch1.parquet").save()
print(f"ID: {artifact.uid}, Version: {artifact.version}")

# From AnnData
adata = ad.read_h5ad("counts.h5ad")
artifact = ln.Artifact.from_anndata(adata, key="scrna/batch1.h5ad", description="scRNA-seq batch 1").save()

# From file path
artifact = ln.Artifact("results/figure.png", key="figures/fig1.png").save()

# Load back
df_loaded = artifact.load()  # Returns DataFrame/AnnData/etc.
path = artifact.cache()       # Returns local file path

# Versioning
artifact_v2 = ln.Artifact.from_df(df_updated, key="experiments/batch1.parquet", revises=artifact).save()
print(f"v1: {artifact.uid}, v2: {artifact_v2.uid}")
print(f"Latest version: {artifact_v2.is_latest}")

# Delete (archive first, then permanent)
artifact.delete(permanent=False)  # Archive
# artifact.delete(permanent=True)  # Permanent deletion

2. 谱系追踪

自动捕获来源信息，确保可复现性。

import lamindb as ln

# Start tracking — captures notebook/script, environment, user
ln.track(params={"method": "PCA", "n_components": 50})

# All artifacts created within this block are linked to this run
input_data = ln.Artifact.get(key="raw/counts.h5ad")
adata = input_data.load()

# ... analysis code ...

output = ln.Artifact.from_anndata(adata, key="processed/pca.h5ad").save()

# View lineage graph
output.view_lineage()

ln.finish()  # Finalize tracking

3. 查询与筛选

按元数据、特征和注释搜索和筛选 artifacts。

import lamindb as ln

# Basic filtering
artifacts = ln.Artifact.filter(key__startswith="scrna/").df()
print(f"Found {len(artifacts)} scRNA-seq artifacts")

# Filter by metadata
recent = ln.Artifact.filter(
    created_at__gte="2026-01-01",
    size__gt=1000000
).df()

# Filter by annotated features
immune = ln.Artifact.filter(
    cell_types__name="T cell",
    tissues__name="PBMC"
).df()

# Single record retrieval
artifact = ln.Artifact.get(key="results/final.parquet")  # Exact match, raises if not found
artifact = ln.Artifact.filter(key="results/final.parquet").one_or_none()  # Returns None if missing

# Full-text search
results = ln.Artifact.search("gene expression PBMC")

# Streaming large files (without full load into memory)
artifact = ln.Artifact.get(key="large_dataset.h5ad")
backed = artifact.open()  # AnnData-backed mode
subset = backed[backed.obs["cell_type"] == "B cell"]

4. 注释与验证

根据架构和本体术语整理数据集。

import lamindb as ln
import bionty as bt

# Annotate artifacts with features
artifact = ln.Artifact.get(key="scrna/batch1.h5ad")
artifact.features.add_values({
    "tissue": "PBMC",
    "condition": "treated",
    "organism": "human",
    "batch": 1
})

# Validate with schema
curator = ln.curators.AnnDataCurator(adata, schema)
try:
    curator.validate()
    artifact = curator.save_artifact(key="validated/batch1.h5ad")
    print("Validation passed")
except ln.errors.ValidationError as e:
    print(f"Validation failed: {e}")

# Standardize cell type names using ontology
adata.obs["cell_type"] = bt.CellType.standardize(adata.obs["cell_type"])

5. 生物学本体（Bionty）

访问标准化的生物学词汇表以进行注释。

import bionty as bt

# Available ontologies
# bt.Gene (Ensembl), bt.Protein (UniProt), bt.CellType (CL),
# bt.Tissue (Uberon), bt.Disease (Mondo), bt.Pathway (GO),
# bt.CellLine (CLO), bt.Phenotype (HPO), bt.Organism (NCBItaxon)

# Import and search ontology
bt.CellType.import_source()
results = bt.CellType.search("T helper")
print(results.head())

# Get specific term
t_cell = bt.CellType.get(name="T cell")
print(f"Ontology ID: {t_cell.ontology_id}")

# Explore hierarchy
children = t_cell.children.all()
parents = t_cell.parents.all()
print(f"Children: {[c.name for c in children]}")

# Validate a list of terms
validated = bt.CellType.validate(["T cell", "B cell", "Unknown_type"])
# Returns boolean array: [True, True, False]

6. 集合与组织

将相关 artifacts 分组，以便批量操作。

import lamindb as ln

# Create a collection
artifacts = ln.Artifact.filter(key__startswith="scrna/batch_").all()
collection = ln.Collection(artifacts, name="scRNA-seq batches Q1 2026").save()
print(f"Collection: {collection.name}, {collection.n_objects} artifacts")

# Query collection
for artifact in collection.artifacts.all():
    print(f"  {artifact.key}: {artifact.size} bytes")

# Organize with hierarchical keys
# Convention: project/experiment/datatype/file
# e.g., "immunology/exp42/scrna/counts.h5ad"

核心概念

核心实体模型

实体	用途	示例
Artifact	带有版本的数据对象	counts.h5ad、results.parquet
Run	单次代码执行	Notebook 运行、脚本执行
Transform	代码定义（Notebook、脚本、工作流）	analysis.ipynb
Feature	类型化元数据字段	tissue、condition、batch
Collection	相关 artifacts 的分组	“实验批次”
ULabel	用于自定义分类的通用标签	“high_quality”、“pilot”

支持的数据类型

格式	方法	使用场景
DataFrame	Artifact.from_df()	表格数据、元数据表
AnnData	Artifact.from_anndata()	单细胞数据
MuData	Artifact.from_mudata()	多模态数据
任意文件	Artifact("path")	图像、FASTQ、自定义格式
Zarr	通过 zarr extra	大型数组数据
TileDB-SOMA	通过 tiledbsoma extra	可扩展的细胞级查询

track() / finish() 模式

每个分析会话都应使用以下方式封装：

ln.track(params={"key": "value"})   # Start: captures code, environment, user
# ... analysis ...
ln.finish()                          # End: finalizes lineage links

常见工作流

工作流：多实验数据湖仓

import lamindb as ln
import anndata as ad

ln.track()

# Register multiple experiments
data_files = ["batch1.h5ad", "batch2.h5ad", "batch3.h5ad"]
tissues = ["PBMC", "bone_marrow", "PBMC"]
conditions = ["control", "treated", "treated"]

for i, (file, tissue, condition) in enumerate(zip(data_files, tissues, conditions)):
    adata = ad.read_h5ad(file)
    artifact = ln.Artifact.from_anndata(
        adata, key=f"scrna/batch_{i}.h5ad", description=f"scRNA-seq batch {i}"
    ).save()
    artifact.features.add_values({
        "tissue": tissue, "condition": condition, "batch": i
    })
    print(f"Registered batch {i}: {artifact.uid}")

# Query across all experiments
treated_pbmc = ln.Artifact.filter(
    key__startswith="scrna/",
    features__tissue="PBMC",
    features__condition="treated"
).all()
print(f"Found {len(treated_pbmc)} matching datasets")

# Load and concatenate
import anndata as ad
adatas = [a.load() for a in treated_pbmc]
combined = ad.concat(adatas)
print(f"Combined: {combined.shape}")

ln.finish()

工作流：经过验证的数据整理

import lamindb as ln
import bionty as bt
import anndata as ad

ln.track()

# 1. Import ontologies
bt.CellType.import_source()
bt.Gene.import_source(organism="human")

# 2. Load raw data
adata = ad.read_h5ad("raw_counts.h5ad")
print(f"Raw: {adata.shape}")

# 3. Validate and standardize cell types
validated = bt.CellType.validate(adata.obs["cell_type"].unique())
if not all(validated):
    adata.obs["cell_type"] = bt.CellType.standardize(adata.obs["cell_type"])

# 4. Validate gene names
gene_validated = bt.Gene.validate(adata.var_names)
print(f"Valid genes: {sum(gene_validated)}/{len(gene_validated)}")

# 5. Curate and save
curator = ln.curators.AnnDataCurator(adata, schema)
curator.validate()
artifact = curator.save_artifact(key="curated/validated_counts.h5ad")
print(f"Saved curated artifact: {artifact.uid}")

ln.finish()

工作流：Nextflow 流水线集成

1. 在每个 Nextflow 进程中导入 lamindb 并调用 ln.track()
2. 使用 ln.Artifact.get(key=...) 加载输入 artifacts；将其缓存到本地路径
3. 运行分析；使用 ln.Artifact(...).save() 将输出保存为新的 artifact
4. 调用 ln.finish() — 谱系会自动关联输入和输出

关键参数

参数	功能	默认值	选项	作用
key	Artifact()	None	字符串路径	分层存储键（例如，“project/data.h5ad”）
description	Artifact()	None	字符串	易于理解的描述
revises	Artifact()	None	Artifact	要修订的上一版本
params	ln.track()	None	Dict	当前运行的参数
organism	bt.Gene.import_source()	None	"human", "mouse"	本体对应的生物
permanent	.delete()	False	True/False	永久删除或归档删除
__startswith	.filter()	—	字符串	键前缀筛选
__gte, __lte	.filter()	—	值	大于等于／小于等于
__contains	.filter()	—	字符串	子字符串匹配

最佳实践

1. 始终使用 ln.track() / ln.finish() 包装分析：这会自动记录血缘信息。否则，工件将没有来源记录。

2. 使用分层键：按 project/experiment/datatype/file.ext 组织（例如，immunology/exp42/scrna/counts.h5ad）。这样可以按前缀查询。

3. 反模式 — 复制数据而非进行版本管理：使用 revises= 参数创建新版本，不要为同一数据集创建新键。

4. 尽早验证：在分析前运行架构验证。及早发现错误元数据可以节省后续调试时间。

5. 使用本体进行标准化：将自由文本标签映射到本体术语（例如，“T helper cell” → CL:0000912）。这样可以跨数据集查询。

6. 反模式 — 未检查大小就加载大型文件：先使用 .filter().df() 检查元数据，再使用 .load() 或 .open()（后备模式）处理大型文件。

7. 先查询元数据，再加载数据：使用 .filter() 筛选出相关工件，然后只加载所需内容。

常见操作示例

操作示例：批量注册数据集

import lamindb as ln
from pathlib import Path

ln.track()

data_dir = Path("raw_data/")
for fcs_file in data_dir.glob("*.fcs"):
    artifact = ln.Artifact(str(fcs_file), key=f"flow_cytometry/{fcs_file.name}").save()
    artifact.features.add_values({"assay": "flow_cytometry", "source": "batch_import"})
    print(f"Registered: {fcs_file.name} -> {artifact.uid}")

ln.finish()

操作示例：查看和导出血缘信息

import lamindb as ln

artifact = ln.Artifact.get(key="results/final_analysis.h5ad")

# View lineage graph (opens in browser or notebook)
artifact.view_lineage()

# Programmatic lineage access
run = artifact.run
print(f"Created by: {run.transform.name}")
print(f"User: {run.created_by.name}")
print(f"Date: {run.created_at}")
print(f"Input artifacts: {[a.key for a in run.input_artifacts.all()]}")

操作示例：探索本体层级

import bionty as bt

bt.CellType.import_source()
t_cell = bt.CellType.get(name="T cell")

# Explore hierarchy
print(f"Parents: {[p.name for p in t_cell.parents.all()]}")
print(f"Children: {[c.name for c in t_cell.children.all()]}")

# Find all descendants
descendants = t_cell.children.all()
for child in descendants:
    grandchildren = child.children.all()
    print(f"  {child.name}: {[gc.name for gc in grandchildren]}")

故障排除

问题	原因	解决方案
InstanceNotSetupError	实例未初始化	运行 lamin init --storage ./data --name my-project
ln.track() 失败	没有转换上下文	在 notebook／脚本中运行，不要在 REPL 中运行；或者显式传入 transform
Artifact key 冲突	键已存在（不是版本）	使用 revises= 进行版本管理，或选择其他键
ValidationError	数据与架构不匹配	运行 curator.validate() 查看具体失败项；标准化术语
大型实例上的查询缓慢	筛选字段没有索引	先使用 .df() 查看概况；为经常筛选的字段添加数据库索引
本体导入失败	网络问题或生物类型错误	检查互联网连接；显式指定 organism="human"
.cache() 出现 FileNotFoundError	云端工件尚未同步	检查存储连接；如需在内存中访问，请改用 artifact.load()

相关技能

• anndata-data-structure — LaminDB 中单细胞数据的主要数据容器格式 AnnData
• scanpy-scrna-seq — 单细胞分析工作流；LaminDB 管理由 scanpy 分析的数据
• scvi-tools-single-cell — 用于单细胞的深度学习模型；与 LaminDB 集成以跟踪数据／模型

参考资料

• LaminDB 文档 — 官方用户指南和 API 参考
• LaminDB 教程 — 分步入门介绍
• Bionty 文档 — 生物本体管理
• LaminDB GitHub — 源代码和问题