Snowpark Python工作表数据完整性校验报错及相关咨询

1. 修复KeyError: 'completeness'错误 #

这个错误通常是因为代码中引用了不存在的字典键或DataFrame列名。比如构建结果字典时键名拼写错误，或者后续访问completeness列时，实际字典里没有这个键。以下是修正后的完整性计算代码：

import snowflake.snowpark as snowpark
from snowflake.snowpark.functions import col, count
import pandas as pd

def main(session: snowpark.Session):
    # 替换为你的目标表名
    sample_df = session.table("YOUR_TABLE").limit(1000)
    total_rows = sample_df.count()
    
    quality_metrics = []
    for col_name in sample_df.columns:
        # 计算非空值数量
        non_null_count = sample_df.select(count(col(col_name))).collect()[0][0]
        completeness = non_null_count / total_rows
        
        # 确保字典键名正确为'completeness'
        quality_metrics.append({
            "column_name": col_name,
            "completeness": round(completeness, 4)
        })
    
    # 转换为Pandas DataFrame展示
    result_df = pd.DataFrame(quality_metrics)
    return result_df

2. 扩展实现完整性、唯一性、最值校验 #

在上述代码基础上，添加唯一性（非空值的唯一占比）、最值的计算逻辑，同时处理数值型与非数值型列的差异：

import snowflake.snowpark as snowpark
from snowflake.snowpark.functions import col, count, count_distinct, min, max
import pandas as pd

def main(session: snowpark.Session):
    sample_df = session.table("YOUR_TABLE").limit(1000)
    total_rows = sample_df.count()
    quality_metrics = []
    
    for col_name in sample_df.columns:
        # 完整性计算
        non_null_count = sample_df.select(count(col(col_name))).collect()[0][0]
        completeness = non_null_count / total_rows if total_rows > 0 else 0
        
        # 唯一性计算（仅针对非空值）
        distinct_count = sample_df.select(count_distinct(col(col_name))).collect()[0][0]
        uniqueness = distinct_count / non_null_count if non_null_count > 0 else 0
        
        # 最值计算（仅针对数值型列）
        col_datatype = str(sample_df.schema[col_name].datatype)
        min_val, max_val = None, None
        if col_datatype.startswith(("NUMBER", "FLOAT", "DOUBLE")):
            min_val = sample_df.select(min(col(col_name))).collect()[0][0]
            max_val = sample_df.select(max(col(col_name))).collect()[0][0]
        
        quality_metrics.append({
            "column_name": col_name,
            "completeness": round(completeness, 4),
            "uniqueness": round(uniqueness, 4),
            "min_value": min_val,
            "max_value": max_val
        })
    
    result_df = pd.DataFrame(quality_metrics)
    return result_df

3. Snowpark是否适合数据探查与质量校验场景 #

完全适合，核心原因包括：

• 直接在Snowflake集群内运行计算，无需将大样本数据拉取到本地，减少数据传输开销，处理大表时效率更高。
• 支持SQL与Python混合开发，之前用SQL实现的校验逻辑可以无缝集成到Python工作流中。
• 利用Snowflake的分布式计算资源，可并行处理多列校验，比本地Python脚本速度更快。
• 可将校验逻辑封装为可复用函数，轻松集成到数据流水线中实现自动化校验。

4. Snowpark中集成Pandas的方法 #

Snowpark提供了多种与Pandas互通的方式：

• Snowpark DataFrame转Pandas DataFrame：使用to_pandas()方法，适合将小样本结果拉到本地做可视化或精细分析，示例：snowpark_df.to_pandas()。
• Pandas DataFrame转Snowpark DataFrame：使用session.create_dataframe(pandas_df)，可将本地Pandas数据上传到Snowflake表中。
• Pandas UDF分布式运行：如果有自定义的Pandas校验逻辑，可注册为UDF，在Snowpark中分布式执行，适合复杂列级校验。
• 注意：转换大表时避免直接使用to_pandas()，会将全量数据拉到本地导致内存溢出，建议先通过limit()取样本，或用Snowpark完成大部分计算后再转Pandas展示结果。

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