cellxgene-census

CZ CELLxGENE Census

概述

CZ CELLxGENE Census 提供对 CZ CELLxGENE Discover 中全面且带版本管理的标准化单细胞基因组学数据集的编程访问能力。此技能可高效查询和分析数千个数据集中的数百万个细胞。

Census 包括：

• 来自人类和小鼠的 6,100 多万个细胞
• 标准化元数据（细胞类型、组织、疾病、供体）
• 原始基因表达矩阵
• 预先计算的嵌入和统计数据
• 与 PyTorch、scanpy 及其他分析工具集成

何时使用此技能

以下情况下应使用此技能：

• 按细胞类型、组织或疾病查询单细胞表达数据
• 探索可用的单细胞数据集和元数据
• 使用单细胞数据训练机器学习模型
• 执行大规模跨数据集分析
• 将 Census 数据与 scanpy 或其他分析框架集成
• 计算数百万个细胞的统计数据
• 访问预先计算的嵌入或模型预测结果

安装和设置

安装 Census API：

uv pip install cellxgene-census

对于机器学习工作流，请安装其他依赖项：

uv pip install cellxgene-census[experimental]

核心工作流模式

1. 打开 Census

始终使用上下文管理器，确保正确清理资源：

import cellxgene_census

# Open latest stable version
with cellxgene_census.open_soma() as census:
    # Work with census data

# Open specific version for reproducibility
with cellxgene_census.open_soma(census_version="2023-07-25") as census:
    # Work with census data

要点：

• 使用上下文管理器（with 语句）自动清理资源
• 指定 census_version，以便进行可复现的分析
• 默认打开最新的“stable”版本

2. 探索 Census 信息

查询表达数据前，请先探索可用的数据集和元数据。

访问摘要信息：

# Get summary statistics
summary = census["census_info"]["summary"].read().concat().to_pandas()
print(f"Total cells: {summary['total_cell_count'][0]}")

# Get all datasets
datasets = census["census_info"]["datasets"].read().concat().to_pandas()

# Filter datasets by criteria
covid_datasets = datasets[datasets["disease"].str.contains("COVID", na=False)]

查询细胞元数据，了解可用数据：

# Get unique cell types in a tissue
cell_metadata = cellxgene_census.get_obs(
    census,
    "homo_sapiens",
    value_filter="tissue_general == 'brain' and is_primary_data == True",
    column_names=["cell_type"]
)
unique_cell_types = cell_metadata["cell_type"].unique()
print(f"Found {len(unique_cell_types)} cell types in brain")

# Count cells by tissue
tissue_counts = cell_metadata.groupby("tissue_general").size()

重要提示： 除非专门分析重复细胞，否则始终筛选 is_primary_data == True，以避免重复计数。

3. 查询表达数据（小到中等规模）

对于返回少于 100k 个细胞且可放入内存的查询，请使用 get_anndata()：

# Basic query with cell type and tissue filters
adata = cellxgene_census.get_anndata(
    census=census,
    organism="Homo sapiens",  # or "Mus musculus"
    obs_value_filter="cell_type == 'B cell' and tissue_general == 'lung' and is_primary_data == True",
    obs_column_names=["assay", "disease", "sex", "donor_id"],
)

# Query specific genes with multiple filters
adata = cellxgene_census.get_anndata(
    census=census,
    organism="Homo sapiens",
    var_value_filter="feature_name in ['CD4', 'CD8A', 'CD19', 'FOXP3']",
    obs_value_filter="cell_type == 'T cell' and disease == 'COVID-19' and is_primary_data == True",
    obs_column_names=["cell_type", "tissue_general", "donor_id"],
)

筛选语法：

• 使用 obs_value_filter 筛选细胞
• 使用 var_value_filter 筛选基因
• 使用 and、or 组合条件
• 使用 in 指定多个值：tissue in ['lung', 'liver']
• 使用 obs_column_names 仅选择所需列

单独获取元数据：

# Query cell metadata
cell_metadata = cellxgene_census.get_obs(
    census, "homo_sapiens",
    value_filter="disease == 'COVID-19' and is_primary_data == True",
    column_names=["cell_type", "tissue_general", "donor_id"]
)

# Query gene metadata
gene_metadata = cellxgene_census.get_var(
    census, "homo_sapiens",
    value_filter="feature_name in ['CD4', 'CD8A']",
    column_names=["feature_id", "feature_name", "feature_length"]
)

4. 大规模查询（外存处理）

对于超过可用 RAM 的查询，请使用 axis_query() 进行迭代处理：

import tiledbsoma as soma

# Create axis query
query = census["census_data"]["homo_sapiens"].axis_query(
    measurement_name="RNA",
    obs_query=soma.AxisQuery(
        value_filter="tissue_general == 'brain' and is_primary_data == True"
    ),
    var_query=soma.AxisQuery(
        value_filter="feature_name in ['FOXP2', 'TBR1', 'SATB2']"
    )
)

# Iterate through expression matrix in chunks
iterator = query.X("raw").tables()
for batch in iterator:
    # batch is a pyarrow.Table with columns:
    # - soma_data: expression value
    # - soma_dim_0: cell (obs) coordinate
    # - soma_dim_1: gene (var) coordinate
    process_batch(batch)

计算增量统计数据：

# Example: Calculate mean expression
n_observations = 0
sum_values = 0.0

iterator = query.X("raw").tables()
for batch in iterator:
    values = batch["soma_data"].to_numpy()
    n_observations += len(values)
    sum_values += values.sum()

mean_expression = sum_values / n_observations

5. 使用 PyTorch 进行机器学习

训练模型时，请使用实验性的 PyTorch 集成：

from cellxgene_census.experimental.ml import experiment_dataloader

with cellxgene_census.open_soma() as census:
    # Create dataloader
    dataloader = experiment_dataloader(
        census["census_data"]["homo_sapiens"],
        measurement_name="RNA",
        X_name="raw",
        obs_value_filter="tissue_general == 'liver' and is_primary_data == True",
        obs_column_names=["cell_type"],
        batch_size=128,
        shuffle=True,
    )

    # Training loop
    for epoch in range(num_epochs):
        for batch in dataloader:
            X = batch["X"]  # Gene expression tensor
            labels = batch["obs"]["cell_type"]  # Cell type labels

            # Forward pass
            outputs = model(X)
            loss = criterion(outputs, labels)

            # Backward pass
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

训练集／测试集划分：

from cellxgene_census.experimental.ml import ExperimentDataset

# Create dataset from experiment
dataset = ExperimentDataset(
    experiment_axis_query,
    layer_name="raw",
    obs_column_names=["cell_type"],
    batch_size=128,
)

# Split into train and test
train_dataset, test_dataset = dataset.random_split(
    split=[0.8, 0.2],
    seed=42
)

6. 与 Scanpy 集成

将 Census 数据无缝集成到 scanpy 工作流中：

import scanpy as sc

# Load data from Census
adata = cellxgene_census.get_anndata(
    census=census,
    organism="Homo sapiens",
    obs_value_filter="cell_type == 'neuron' and tissue_general == 'cortex' and is_primary_data == True",
)

# Standard scanpy workflow
sc.pp.normalize_total(adata, target_sum=1e4)
sc.pp.log1p(adata)
sc.pp.highly_variable_genes(adata, n_top_genes=2000)

# Dimensionality reduction
sc.pp.pca(adata, n_comps=50)
sc.pp.neighbors(adata)
sc.tl.umap(adata)

# Visualization
sc.pl.umap(adata, color=["cell_type", "tissue", "disease"])

7. 多数据集集成

查询并集成多个数据集：

# Strategy 1: Query multiple tissues separately
tissues = ["lung", "liver", "kidney"]
adatas = []

for tissue in tissues:
    adata = cellxgene_census.get_anndata(
        census=census,
        organism="Homo sapiens",
        obs_value_filter=f"tissue_general == '{tissue}' and is_primary_data == True",
    )
    adata.obs["tissue"] = tissue
    adatas.append(adata)

# Concatenate
combined = adatas[0].concatenate(adatas[1:])

# Strategy 2: Query multiple datasets directly
adata = cellxgene_census.get_anndata(
    census=census,
    organism="Homo sapiens",
    obs_value_filter="tissue_general in ['lung', 'liver', 'kidney'] and is_primary_data == True",
)

核心概念和最佳实践

始终筛选主要数据

除非要分析重复细胞，否则查询时始终包含 is_primary_data == True，以避免重复计数：

obs_value_filter="cell_type == 'B cell' and is_primary_data == True"

指定 Census 版本以确保可复现性

在生产分析中始终指定 Census 版本：

census = cellxgene_census.open_soma(census_version="2023-07-25")

加载前估算查询规模

对于大型查询，请先检查细胞数量，以避免内存问题：

# Get cell count
metadata = cellxgene_census.get_obs(
    census, "homo_sapiens",
    value_filter="tissue_general == 'brain' and is_primary_data == True",
    column_names=["soma_joinid"]
)
n_cells = len(metadata)
print(f"Query will return {n_cells:,} cells")

# If too large (>100k), use out-of-core processing

使用 tissue_general 进行较宽泛的分类

tissue_general 字段提供比 tissue 更粗略的类别，适用于跨组织分析：

# Broader grouping
obs_value_filter="tissue_general == 'immune system'"

# Specific tissue
obs_value_filter="tissue == 'peripheral blood mononuclear cell'"

仅选择所需列

仅指定所需的元数据列，以尽量减少数据传输：

obs_column_names=["cell_type", "tissue_general", "disease"]  # Not all columns

检查基因特定查询中的数据集是否包含目标基因

分析特定基因时，请确认哪些数据集测量了这些基因：

presence = cellxgene_census.get_presence_matrix(
    census,
    "homo_sapiens",
    var_value_filter="feature_name in ['CD4', 'CD8A']"
)

两步工作流：先探索，再查询

先探索元数据以了解可用数据，再查询表达数据：

# Step 1: Explore what's available
metadata = cellxgene_census.get_obs(
    census, "homo_sapiens",
    value_filter="disease == 'COVID-19' and is_primary_data == True",
    column_names=["cell_type", "tissue_general"]
)
print(metadata.value_counts())

# Step 2: Query based on findings
adata = cellxgene_census.get_anndata(
    census=census,
    organism="Homo sapiens",
    obs_value_filter="disease == 'COVID-19' and cell_type == 'T cell' and is_primary_data == True",
)

可用元数据字段

细胞元数据（obs）

筛选用的关键字段：

• cell_type、cell_type_ontology_term_id
• tissue、tissue_general、tissue_ontology_term_id
• disease、disease_ontology_term_id
• assay、assay_ontology_term_id
• donor_id、sex、self_reported_ethnicity
• development_stage、development_stage_ontology_term_id
• dataset_id
• is_primary_data（布尔值：True = 唯一细胞）

基因元数据（var）

• feature_id（Ensembl 基因 ID，例如 "ENSG00000161798"）
• feature_name（基因符号，例如 "FOXP2"）
• feature_length（基因长度，单位为碱基对）

参考文档

此技能包含详尽的参考文档：

references/census_schema.md

全面介绍：

• Census 数据的结构与组织方式
• 所有可用的元数据字段
• 值筛选语法和运算符
• SOMA 对象类型
• 数据纳入标准

阅读时机： 需要详细了解架构信息、完整的元数据字段列表或复杂的筛选语法时。

references/common_patterns.md

示例和模式涵盖：

• 探索性查询（仅元数据）
• 中小型查询（AnnData）
• 大型查询（外存处理）
• PyTorch 集成
• Scanpy 集成工作流
• 多数据集集成
• 最佳实践和常见陷阱

阅读时机： 实现特定查询模式、查找代码示例或排查常见问题时。

常见用例

用例 1：探索组织中的细胞类型

with cellxgene_census.open_soma() as census:
    cells = cellxgene_census.get_obs(
        census, "homo_sapiens",
        value_filter="tissue_general == 'lung' and is_primary_data == True",
        column_names=["cell_type"]
    )
    print(cells["cell_type"].value_counts())

用例 2：查询标记基因表达

with cellxgene_census.open_soma() as census:
    adata = cellxgene_census.get_anndata(
        census=census,
        organism="Homo sapiens",
        var_value_filter="feature_name in ['CD4', 'CD8A', 'CD19']",
        obs_value_filter="cell_type in ['T cell', 'B cell'] and is_primary_data == True",
    )

用例 3：训练细胞类型分类器

from cellxgene_census.experimental.ml import experiment_dataloader

with cellxgene_census.open_soma() as census:
    dataloader = experiment_dataloader(
        census["census_data"]["homo_sapiens"],
        measurement_name="RNA",
        X_name="raw",
        obs_value_filter="is_primary_data == True",
        obs_column_names=["cell_type"],
        batch_size=128,
        shuffle=True,
    )

    # Train model
    for epoch in range(epochs):
        for batch in dataloader:
            # Training logic
            pass

用例 4：跨组织分析

with cellxgene_census.open_soma() as census:
    adata = cellxgene_census.get_anndata(
        census=census,
        organism="Homo sapiens",
        obs_value_filter="cell_type == 'macrophage' and tissue_general in ['lung', 'liver', 'brain'] and is_primary_data == True",
    )

    # Analyze macrophage differences across tissues
    sc.tl.rank_genes_groups(adata, groupby="tissue_general")

故障排查

查询返回的细胞过多

• 添加更具体的筛选条件以缩小范围
• 使用 tissue 而非 tissue_general，以获得更细的粒度
• 如果已知具体的 dataset_id，则按其筛选
• 对大型查询改用外存处理

内存错误

• 使用更严格的筛选条件缩小查询范围
• 使用 var_value_filter 选择较少的基因
• 使用 axis_query() 进行外存处理
• 分批处理数据

结果中存在重复细胞

• 始终在筛选条件中包含 is_primary_data == True
• 检查是否有意跨多个数据集进行查询

找不到基因

• 确认基因名称拼写正确（区分大小写）
• 尝试使用 feature_id 查询 Ensembl ID，而非 feature_name
• 检查数据集存在矩阵，确认是否测量了该基因
• 在构建 Census 时，某些基因可能已被筛除

版本不一致

• 始终明确指定 census_version
• 所有分析中使用相同版本
• 查看发行说明，了解特定版本的变更