frompyspark.sqlimportSparkSessionfrompyspark.sql.functionsimportcol,coalesce,trim,when,lit,sumfrompyspark.sql.typesimportStringType,NumericType# 在 Databricks 中spark 会话通常已经存在无需重新创建# 如果需要显式创建使用# spark SparkSession.builder.getOrCreate()# 配置参数database_nameyour_database# 替换为实际数据库名result_list[]# 获取数据库中的所有表和视图包括 Delta 表tablesspark.catalog.listTables(database_name)fortableintables:table_nametable.name full_table_namef{database_name}.{table_name}try:# 读取表或视图Delta 表和普通表均可dfspark.table(full_table_name)# 快速判断是否为空表避免不必要的缓存total_countdf.count()iftotal_count0:continue# 缓存数据以便多次引用实际上我们只需扫描一次缓存并非必需df.cache()# 构建所有字段的聚合表达式一次性计算agg_exprs[]field_meta[]# 用于记录每个字段的类型和名称forfieldindf.schema.fields:col_namefield.name col_typefield.dataTypeifisinstance(col_type,StringType):# 字符串类型统计 null 或 trim 后为空字符串modified_coltrim(coalesce(col(col_name),lit()))condition(modified_collit())count_exprsum(when(condition,1).otherwise(0)).alias(fcnt_{col_name})elifisinstance(col_type,NumericType):# 数值类型统计 null 或零值modified_colcoalesce(col(col_name),lit(0))condition(modified_collit(0))count_exprsum(when(condition,1).otherwise(0)).alias(fcnt_{col_name})else:# 其他类型仅统计 nullconditioncol(col_name).isNull()count_exprsum(when(condition,1).otherwise(0)).alias(fcnt_{col_name})agg_exprs.append(count_expr)field_meta.append((col_name,str(col_type)))# 执行一次聚合获取所有字段的统计值stats_rowdf.agg(*agg_exprs).collect()[0]# 整理结果forcol_name,col_typeinfield_meta:stat_countstats_row[fcnt_{col_name}]percentageround((stat_count/total_count)*100,2)iftotal_count0else0.0result_list.append((database_name,table_name,col_name,col_type,stat_count,total_count,float(percentage)))df.unpersist()# 释放缓存exceptExceptionase:print(fError processing table{table_name}:{str(e)})continue# 创建结果 DataFrameresult_columns[database_name,table_name,column_name,column_type,stat_count,total_rows,percentage]result_dfspark.createDataFrame(result_list,result_columns)# 显示结果result_df.show(truncateFalse)# 可选将结果保存到 Delta 表# result_df.write.format(delta).mode(overwrite).saveAsTable(your_audit_table)主要改动说明原代码修改后原因显式创建SparkSession并启用 Hive直接使用 Databricks 内置sparkDatabricks 已预配置无需额外初始化对每个字段分别执行df.agg()构建所有字段聚合表达式一次df.agg()减少表扫描次数大幅提高性能缓存df后多次扫描缓存后仅一次聚合扫描配合优化降低缓存开销spark.catalog.listTables()同样使用能返回表和视图无需修改原生支持 Delta 表和视图异常处理与结果收集保持不变逻辑通用注意事项数据库名称根据实际环境修改database_name如果使用默认数据库可设为default。视图spark.table()可以读取持久化视图但临时视图不会出现在listTables中。如需处理临时视图可额外指定名称列表。性能对于大型表df.count()和聚合仍会触发一次完整扫描这是必要的。如果表极大可考虑分批处理或使用抽样估算。Delta 表Delta 表在 Catalog 中通常显示为普通表名读取方式与 Hive 表一致无需特殊处理。
frompyspark.sqlimportSparkSessionfrompyspark.sql.functionsimportcol,coalesce,trim,when,lit,sumfrompyspark.sql.typesimportStringType,NumericType# 在 Databricks 中spark 会话通常已经存在无需重新创建# 如果需要显式创建使用# spark SparkSession.builder.getOrCreate()# 配置参数database_nameyour_database# 替换为实际数据库名result_list[]# 获取数据库中的所有表和视图包括 Delta 表tablesspark.catalog.listTables(database_name)fortableintables:table_nametable.name full_table_namef{database_name}.{table_name}try:# 读取表或视图Delta 表和普通表均可dfspark.table(full_table_name)# 快速判断是否为空表避免不必要的缓存total_countdf.count()iftotal_count0:continue# 缓存数据以便多次引用实际上我们只需扫描一次缓存并非必需df.cache()# 构建所有字段的聚合表达式一次性计算agg_exprs[]field_meta[]# 用于记录每个字段的类型和名称forfieldindf.schema.fields:col_namefield.name col_typefield.dataTypeifisinstance(col_type,StringType):# 字符串类型统计 null 或 trim 后为空字符串modified_coltrim(coalesce(col(col_name),lit()))condition(modified_collit())count_exprsum(when(condition,1).otherwise(0)).alias(fcnt_{col_name})elifisinstance(col_type,NumericType):# 数值类型统计 null 或零值modified_colcoalesce(col(col_name),lit(0))condition(modified_collit(0))count_exprsum(when(condition,1).otherwise(0)).alias(fcnt_{col_name})else:# 其他类型仅统计 nullconditioncol(col_name).isNull()count_exprsum(when(condition,1).otherwise(0)).alias(fcnt_{col_name})agg_exprs.append(count_expr)field_meta.append((col_name,str(col_type)))# 执行一次聚合获取所有字段的统计值stats_rowdf.agg(*agg_exprs).collect()[0]# 整理结果forcol_name,col_typeinfield_meta:stat_countstats_row[fcnt_{col_name}]percentageround((stat_count/total_count)*100,2)iftotal_count0else0.0result_list.append((database_name,table_name,col_name,col_type,stat_count,total_count,float(percentage)))df.unpersist()# 释放缓存exceptExceptionase:print(fError processing table{table_name}:{str(e)})continue# 创建结果 DataFrameresult_columns[database_name,table_name,column_name,column_type,stat_count,total_rows,percentage]result_dfspark.createDataFrame(result_list,result_columns)# 显示结果result_df.show(truncateFalse)# 可选将结果保存到 Delta 表# result_df.write.format(delta).mode(overwrite).saveAsTable(your_audit_table)主要改动说明原代码修改后原因显式创建SparkSession并启用 Hive直接使用 Databricks 内置sparkDatabricks 已预配置无需额外初始化对每个字段分别执行df.agg()构建所有字段聚合表达式一次df.agg()减少表扫描次数大幅提高性能缓存df后多次扫描缓存后仅一次聚合扫描配合优化降低缓存开销spark.catalog.listTables()同样使用能返回表和视图无需修改原生支持 Delta 表和视图异常处理与结果收集保持不变逻辑通用注意事项数据库名称根据实际环境修改database_name如果使用默认数据库可设为default。视图spark.table()可以读取持久化视图但临时视图不会出现在listTables中。如需处理临时视图可额外指定名称列表。性能对于大型表df.count()和聚合仍会触发一次完整扫描这是必要的。如果表极大可考虑分批处理或使用抽样估算。Delta 表Delta 表在 Catalog 中通常显示为普通表名读取方式与 Hive 表一致无需特殊处理。
