如何在Databricks中对表的大量列子集应用过滤条件？

需要对user_STE中由column_list定义的400+列，筛选出至少2列满足值>0且不为NULL的记录，原SQL因无法动态引用列名导致失效，以下是两种可行方案：

方案一：动态SQL生成（适合纯SQL场景） #

思路是先从column_list获取目标列名，动态拼接出每个列的判断逻辑，通过统计符合条件的列数完成过滤。

1. 先获取目标列的字符串列表：

WITH column_list AS (
    SELECT column_name
    FROM information_schema.columns
    WHERE table_name = 'ccdc_sustained_member_subs_engagement_pivot'
      AND table_schema = 'ccea_prod'
      AND column_name NOT LIKE '%ACROBAT%'
      AND column_name LIKE '%active_days%'
)
SELECT STRING_AGG(column_name, ',') AS target_columns
FROM column_list;

执行后会得到类似col1,col2,col3,...的列名字符串。

2. 拼接并执行最终过滤SQL：
   将上述得到的列名替换到下方的列位置，生成完整查询：

WITH user_STE AS (
    SELECT *
    FROM ccea_prod.ccdc_sustained_member_subs_engagement_pivot
    WHERE market_area = 'US'
      AND period_end_date = '2024-09-06'
      AND market_segment = 'EDUCATION'
      AND subscription_offerings = 'STE - All Apps'
      AND subscription_type = 'IN'
)
SELECT *
FROM user_STE
WHERE (
    CASE WHEN app1_active_days > 0 AND app1_active_days IS NOT NULL THEN 1 ELSE 0 END +
    CASE WHEN app2_active_days > 0 AND app2_active_days IS NOT NULL THEN 1 ELSE 0 END +
    -- ... 依次添加所有目标列的CASE语句
) >= 2;

如果想自动化拼接，可使用Databricks Python代码生成SQL：

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# 获取目标列名
column_list_df = spark.sql("""
    SELECT column_name
    FROM information_schema.columns
    WHERE table_name = 'ccdc_sustained_member_subs_engagement_pivot'
      AND table_schema = 'ccea_prod'
      AND column_name NOT LIKE '%ACROBAT%'
      AND column_name LIKE '%active_days%'
""")
target_columns = [row.column_name for row in column_list_df.collect()]

# 生成CASE语句片段
case_statements = " + ".join([f"CASE WHEN {col} > 0 AND {col} IS NOT NULL THEN 1 ELSE 0 END" for col in target_columns])

# 拼接完整SQL并执行
final_sql = f"""
WITH user_STE AS (
    SELECT *
    FROM ccea_prod.ccdc_sustained_member_subs_engagement_pivot
    WHERE market_area = 'US'
      AND period_end_date = '2024-09-06'
      AND market_segment = 'EDUCATION'
      AND subscription_offerings = 'STE - All Apps'
      AND subscription_type = 'IN'
)
SELECT *
FROM user_STE
WHERE ({case_statements}) >= 2
"""

result_df = spark.sql(final_sql)
result_df.show()

方案二：使用Spark SQL内置函数（简洁高效） #

利用Spark的array函数将目标列转为数组，再用filter筛选符合条件的元素，最后通过size判断数量是否达标：

WITH column_list AS (
    SELECT collect_list(column_name) AS target_columns
    FROM information_schema.columns
    WHERE table_name = 'ccdc_sustained_member_subs_engagement_pivot'
      AND table_schema = 'ccea_prod'
      AND column_name NOT LIKE '%ACROBAT%'
      AND column_name LIKE '%active_days%'
),
user_STE AS (
    SELECT *,
           size(filter(array({(SELECT target_columns FROM column_list)}), x -> x > 0 AND x IS NOT NULL)) AS valid_col_count
    FROM ccea_prod.ccdc_sustained_member_subs_engagement_pivot
    WHERE market_area = 'US'
      AND period_end_date = '2024-09-06'
      AND market_segment = 'EDUCATION'
      AND subscription_offerings = 'STE - All Apps'
      AND subscription_type = 'IN'
)
SELECT * EXCEPT(valid_col_count)
FROM user_STE
WHERE valid_col_count >= 2;

注：{(SELECT target_columns FROM column_list)}会自动将收集到的列名展开为col1, col2, col3,...，直接嵌入array函数中。

原代码失效原因 #

原SQL中的COALESCE(NULLIF(user_STE.[column_name], 0), NULL) IS NOT NULL无法生效，因为column_name是字符串类型的列名，静态SQL无法将其解析为user_STE的实际列，必须通过动态生成列引用或数组转换的方式处理。

内容的提问来源于stack exchange，提问作者Hctor Alonso Hormazbal Vildsol