Polars 加列时的 Schema 常见坑与修复思路，外加 GitHub 用户贡献速查技巧

预计阅读时间：11 分钟

数据管道跑着跑着，突然因为一列类型不对就崩了——用 Polars 的人多半踩过这个坑。与此同时，开源协作中你经常需要快速判断一个 GitHub 账号值不值得信任：他是活跃贡献者还是机器人？本文把这两个实用话题拆开讲，各给一套可直接跑的代码。

Polars 管道里的 Schema 问题从哪来

Polars 是强类型的。一个 DataFrame 的 Schema 在创建时就锁定了每列的类型。这带来性能优势，但也意味着：加列时类型不匹配，不会像 pandas 那样悄悄容忍，而是直接报错或产生意料之外的类型。

典型场景有三类：

1. 新列类型和预期不一致

import polars as pl

df = pl.DataFrame({"user_id": [1, 2, 3], "score": [85, 90, 78]})

# 想加一列 "level"，用 with_columns
df = df.with_columns(
    pl.when(pl.col("score") >= 90)
    .then(pl.lit("A"))
    .otherwise(pl.lit("B"))
    .alias("level")
)

print(df.schema)  # {'user_id': Int64, 'score': Int64, 'level': String}

看起来没问题，但如果后续管道期望 level 是 Categorical 类型，下游聚合或 join 可能行为不同。

2. 空值导致类型漂移

df2 = pl.DataFrame({"name": ["alice", "bob", None]})

# 加一列全是 null 的列——类型会变成 Null
df2 = df2.with_columns(pl.lit(None).alias("tag"))

print(df2.schema)  # {'name': String, 'tag': Null}

Null 类型几乎无法和任何列 join 或运算。管道下游一旦碰上就会报错。

3. 多源拼接时列名相同但类型不同

两个 DataFrame 都有 amount 列，一个是 Int64，另一个是 Float64。pl.concat 默认按位置拼接，类型冲突会直接抛异常。

Schema 问题的几种防御姿势

显式声明 Schema，不让类型靠推断

创建 DataFrame 时直接锁定类型：

schema = {
    "user_id": pl.Int64,
    "score": pl.Int64,
    "level": pl.Categorical,
}

df = pl.DataFrame(
    {"user_id": [1, 2, 3], "score": [85, 90, 78], "level": ["B", "A", "B"]},
    schema=schema,
)

加列时强制 cast

df = df.with_columns(
    pl.when(pl.col("score") >= 90)
    .then(pl.lit("A"))
    .otherwise(pl.lit("B"))
    .cast(pl.Categorical)
    .alias("level")
)

给空列指定类型

df2 = df2.with_columns(
    pl.lit(None).cast(pl.String).alias("tag")
)
print(df2.schema)  # {'name': String, 'tag': String}

concat 前统一类型

df_a = pl.DataFrame({"amount": [100, 200]}, schema={"amount": pl.Int64})
df_b = pl.DataFrame({"amount": [50.5, 75.0]}, schema={"amount": pl.Float64})

# 先把 df_a 的 amount cast 成 Float64 再拼接
df_a = df_a.with_columns(pl.col("amount").cast(pl.Float64))
result = pl.concat([df_a, df_b])

管道入口做 Schema 校验

在管道关键节点插入一个校验函数，类型不对就提前报错，而不是跑到下游才崩：

EXPECTED_SCHEMA = {
    "user_id": pl.Int64,
    "score": pl.Int64,
    "level": pl.Categorical,
}

def validate_schema(df: pl.DataFrame, expected: dict) -> pl.DataFrame:
    actual = df.schema
    mismatches = {
        col: (actual.get(col), expected[col])
        for col in expected
        if actual.get(col) != expected[col]
    }
    if mismatches:
        raise ValueError(f"Schema mismatch: {mismatches}")
    return df

# 在管道中使用
df = validate_schema(df, EXPECTED_SCHEMA)

这套做法的好处：错误信息精确到列和类型，排查时间从"下游报错倒推"变成"入口直接定位"。

GitHub 用户贡献速查：快速判断一个账号的投入程度

开源项目中，你经常需要评估一个贡献者：PR 是不是机器生成的？Issue 是不是模板刷的？手动翻 GitHub 页面太慢，用 API 可以几秒出结论。

最小化画像脚本

以下脚本输入一个 GitHub 用户名，输出关键指标：

import requests
import sys

def profile_github_user(username: str, token: str = "") -> dict:
    headers = {"Accept": "application/vnd.github+json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"

    # 用户基本信息
    user_resp = requests.get(f"https://api.github.com/users/{username}", headers=headers)
    user_resp.raise_for_status()
    user = user_resp.json()

    # 公开事件（最近 30 条）
    events_resp = requests.get(
        f"https://api.github.com/users/{username}/events/public?per_page=30",
        headers=headers,
    )
    events_resp.raise_for_status()
    events = events_resp.json()

    # 统计事件类型分布
    event_types = {}
    for e in events:
        t = e.get("type", "Unknown")
        event_types[t] = event_types.get(t, 0) + 1

    # 统计最近操作的仓库数
    repos_touched = set()
    for e in events:
        repo_name = e.get("repo", {}).get("name", "")
        if repo_name:
            repos_touched.add(repo_name)

    return {
        "username": username,
        "public_repos": user.get("public_repos", 0),
        "followers": user.get("followers", 0),
        "account_age_days": (
            (None)  # 需要日期解析，简化处理
            if not user.get("created_at")
            else None  # 实际可用 datetime 解析
        ),
        "bio": user.get("bio", ""),
        "event_type_distribution": event_types,
        "repos_touched_recently": len(repos_touched),
        "recent_repo_list": sorted(repos_touched)[:5],
    }

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python gh_profile.py <username> [token]")
        sys.exit(1)

    uname = sys.argv[1]
    tok = sys.argv[2] if len(sys.argv) > 2 else ""
    result = profile_github_user(uname, tok)

    print(f"=== GitHub User: {result['username']} ===")
    print(f"Public repos:    {result['public_repos']}")
    print(f"Followers:       {result['followers']}")
    print(f"Bio:             {result['bio'] or '(empty)'}")
    print(f"Recent activity across {result['repos_touched_recently']} repos")
    print(f"Top repos:       {result['recent_repo_list']}")
    print(f"Event distribution:")
    for etype, count in sorted(
        result["event_type_distribution"].items(), key=lambda x: -x[1]
    ):
        print(f"  {etype}: {count}")

运行方式：

# 无 token（受 API 60次/小时限制）
python gh_profile.py torvalds

# 有 token（5000次/小时）
python gh_profile.py torvalds ghp_YourTokenHere

怎么解读输出

几个快速判断规则：

信号	可能含义
`PushEvent` 占绝大多数	真正在写代码的人
`IssueCommentEvent` 多但 `PushEvent` 极少	多是讨论者而非代码贡献者
`WatchEvent` / `ForkEvent` 占大头	低参与度，可能只是围观
30 条事件里涉及 1 个仓库	专注单一项目，深度贡献可能性高
30 条事件里涉及 20+ 仓库	广度大但深度可能有限
`bio` 为空 + `public_repos` 为 0 + 事件全是 `ForkEvent`	大概率是机器人或低价值账号

这些不是绝对判断，但能帮你把"手动翻页面 10 分钟"压缩到"跑脚本 3 秒 + 看输出 30 秒"。

实战组合：两个技巧的交叉场景

假设你维护一个开源项目的贡献者仪表盘，数据管道用 Polars 处理贡献者数据，同时用 GitHub API 定期拉取画像。Schema 校验在这里尤其重要——API 字段可能缺失或类型变化。

import polars as pl

# 模拟从多个 GitHub 用户画像汇总的数据
raw_data = [
    {"username": "alice", "public_repos": 12, "followers": 340, "push_ratio": 0.7},
    {"username": "bob", "public_repos": 3, "followers": 5, "push_ratio": 0.05},
    {"username": "bot_x", "public_repos": 0, "followers": 0, "push_ratio": None},
]

CONTRIBUTOR_SCHEMA = {
    "username": pl.String,
    "public_repos": pl.Int64,
    "followers": pl.Int64,
    "push_ratio": pl.Float64,
}

df = pl.DataFrame(raw_data, schema=CONTRIBUTOR_SCHEMA)

# push_ratio 为 null 的行——大概率是低价值账号
df = df.with_columns(
    pl.when(pl.col("push_ratio").is_null())
    .then(pl.lit("low"))
    .when(pl.col("push_ratio") >= 0.5)
    .then(pl.lit("high"))
    .otherwise(pl.lit("medium"))
    .cast(pl.Categorical)
    .alias("trust_level")
)

print(df)

输出：

shape: (3, 5)
┌──────────┬──────────────┬───────────┬────────────┬────────────┐
│ username ┆ public_repos ┆ followers ┆ push_ratio ┆ trust_level │
│ ---      ┆ ---          ┆ ---       ┆ ---        ┆ ---        │
│ str      ┆ i64          ┆ i64       ┆ f64        ┆ cat        │
╞══════════╡══════════════╡═══════════╡════════════╡════════════╡
│ alice    ┆ 12           ┆ 340       ┆ 0.7        ┆ high       │
│ bob      ┆ 3            ┆ 5         ┆ 0.05       ┆ medium     │
│ bot_x    ┆ 0            ┆ 0         ┆ null       ┆ low        │
└──────────┴──────────────┴───────────┴────────────┴────────────┘

注意 push_ratio 列有 null，但 Schema 声明为 Float64，Polars 能正确处理——前提是你在创建 DataFrame 时显式指定了 Schema。如果靠自动推断，None 值会让类型变成 Null，后续 .is_null() 还能工作，但任何算术运算都会报错。

采纳建议与注意事项

Polars Schema 方面：

在管道入口和每个关键转换节点做 Schema 校验，比在下游排查省几小时。
with_columns 加列时养成 cast 习惯，尤其是 Categorical、Datetime 这类对下游行为敏感的类型。
pl.lit(None) 一定要跟 .cast(目标类型) 配合使用，否则你会得到 Null 类型列。
多源 concat 前打印 df.schema 对比，类型不一致就提前 cast。

GitHub 画像方面：

无 token 时 GitHub API 限速 60 次/小时，批量跑一定要带 token。
events/public 只返回最近事件，不代表用户历史全貌；对长期贡献评估还需结合 repos API 看 commit 历史。
事件类型分布是快速信号，不是最终判断。一个账号可能最近在休假，PushEvent 为零但历史贡献很高。
注意 GitHub API 的分页，上面脚本只取了前 30 条事件，生产环境需要循环翻页。

两个技巧看似无关，但在"数据管道 + 外部数据源"的组合场景里，Schema 管理和外部 API 数据的质量控制是同一件事的两面：管道内部靠 Schema 校验保类型一致，管道入口靠 API 数据理解保语义正确。