如何构建不会因类型问题中断运行的Polars动态表达式

问题场景 #

动态批量对Polars全列执行通用表达式时，会因列数据类型不兼容触发运行错误，手动排除不兼容列的方案灵活性差。目标是实现类型不匹配时不中断流程，对执行失败的列返回None、-1等自定义默认值。

示例数据从Parquet文件读取，Schema如下：

import polars as pl

df = pl.scan_parquet("/path/to/data/*.parquet")

print(df.schema)
# 输出: Schema([('date_time', Datetime),('incident', String), ('address', String), ('city', String), ('zipcode', Int32)])

典型报错场景：

# 统计所有列中大于0的值的数量
# 报错: exceptions.ComputeError: 无法将字符串类型与数值类型进行比较
print(df.select((pl.all() > 0).count().name.prefix("__positive_count_")).collect())


# 统计所有列的唯一值计数
# 报错: pyo3_runtime.PanicException: datetime[ns] 类型未实现'unique_counts'方法
print(df.select(pl.all().unique_counts().name.prefix("__unique_count_")).collect())


# 统计所有列中长度大于0的字符串数量
# 报错: exceptions.SchemaError: Series 数据类型 Int32 与 String 不匹配
print(df.select((pl.all().str.len_chars() > 0).count().name.prefix("__empty_count_")).collect())

最优实现方案 #

方案1：基于Schema静态匹配表达式（性能优先，生产环境推荐） #

Polars Lazy模式在执行前即可拿到完整Schema，无需等到运行时就能判断每列的类型兼容性，提前为不同类型列分配对应逻辑，无额外运行时开销。

• 对支持目标操作的列，正常执行表达式
• 对不支持操作的列，直接返回预设默认值

实现示例：

from polars.selectors import numeric, string

# 1. 统计大于0的值数量：仅数值列执行比较逻辑，其余列返回默认值-1
positive_count_exprs = [
    (numeric() > 0).count().name.prefix("__positive_count_"),
    *[pl.lit(-1).alias(f"__positive_count_{col}") for col, dtype in df.schema.items() if not dtype.is_numeric()]
]

# 2. 统计唯一值计数：跳过不支持unique_counts的时间类型列，返回None
unsupported_for_unique = [pl.Datetime, pl.Date, pl.Time]
unique_count_exprs = [
    *[pl.col(col).unique_counts().alias(f"__unique_count_{col}") for col, dtype in df.schema.items() if dtype not in unsupported_for_unique],
    *[pl.lit(None).alias(f"__unique_count_{col}") for col, dtype in df.schema.items() if dtype in unsupported_for_unique]
]

# 3. 统计非空字符串长度：所有列先安全转字符串再计算，从根源避免类型错误
empty_count_exprs = (
    pl.all().cast(pl.String, strict=False).str.len_chars().gt(0).count()
    .name.prefix("__empty_count_")
]

# 统一执行，无类型报错
result = df.select(*positive_count_exprs, *unique_count_exprs, empty_count_exprs).collect()

如果需要频繁写安全表达式，可以封装通用工具函数减少重复代码：

def build_safe_exprs(expr_func, default_val=-1, is_supported=None):
    """
    构建安全的逐列表达式
    :param expr_func: 输入列名，返回对应列的Polars表达式
    :param default_val: 不支持的列返回的默认值
    :param is_supported: 输入dtype，返回布尔值表示该类型是否支持当前操作
    """
    exprs = []
    for col_name, dtype in df.schema.items():
        if is_supported(dtype):
            exprs.append(expr_func(col_name).alias(col_name))
        else:
            exprs.append(pl.lit(default_val).alias(col_name))
    return exprs

方案2：表达式内置错误捕获（开发效率优先） #

Polars 0.20.31及以上版本提供catch_errors方法，可以直接捕获单条表达式的执行错误，失败时返回指定默认值，写法更简洁，适合快速开发场景，性能略低于静态匹配方案。

示例：

# 执行unique_counts失败时返回None
safe_unique_expr = pl.all().unique_counts().catch_errors(return_value=pl.lit(None)).name.prefix("__unique_count_")
print(df.select(safe_unique_expr).collect())

注意：catch_errors仅能捕获表达式执行阶段的错误，Schema解析阶段的类型错误仍需通过类型转换或静态匹配规避。

内容的提问来源于stack exchange，提问作者Chitral Verma