寻求高效SQL实现：保留(B,C)组合在全表合并后出现≥10次的行

步骤1：全局统计(B,C)的出现次数 #

先遍历所有100张表，仅提取(B,C)字段做计数，避免加载冗余的A字段，大幅减少IO和内存占用：

// Scala 示例，Spark环境
import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("BCPairFilter").getOrCreate()

// 读取所有表（支持路径通配符，比如hdfs://data/tables/tbl_* 或Hive表名前缀）
val allBCPairs = spark.read
  .option("header", "true") // 按实际表结构调整是否需要表头
  .csv("hdfs://data/tables/tbl_*")
  .select("B", "C")

// 计算全局出现次数，筛选出≥10次的(B,C)组合
val validBCPairs = allBCPairs
  .groupBy("B", "C")
  .count()
  .filter("count >= 10")
  .select("B", "C") // 只保留核心字段，缩小后续关联的数据量

步骤2：过滤每张表的有效行 #

根据集群资源情况，选择两种实现方式：

方式一：批量处理（资源充足时优先）

一次性读取所有表，与validBCPairs做内关联，直接保留符合条件的行，再按需写回：

val allTablesData = spark.read
  .option("header", "true")
  .csv("hdfs://data/tables/tbl_*")

val filteredData = allTablesData
  .join(validBCPairs, Seq("B", "C"), "inner")

// 若需要按原表结构拆分存储，可读取时保留表源信息再分区写入
filteredData.write
  .option("header", "true")
  .csv("hdfs://data/filtered_tables/")

方式二：逐表处理（内存有限时使用）

逐个读取单表，关联缓存后的validBCPairs，处理完成后写回，避免一次性加载全量数据：

// 提前准备所有表的路径列表
val tablePaths = List("hdfs://data/tables/tbl_1", "hdfs://data/tables/tbl_2", ..., "hdfs://data/tables/tbl_100")

// 缓存validBCPairs，避免多次关联重复计算
validBCPairs.cache()

tablePaths.foreach { path =>
  val singleTable = spark.read.option("header", "true").csv(path)
  val filteredTable = singleTable.join(validBCPairs, Seq("B", "C"), "inner")
  filteredTable.write.option("header", "true").csv(s"hdfs://data/filtered_tables/${path.split("/").last}")
}

validBCPairs.unpersist()

关键优化点 #

• 字段裁剪：统计阶段只加载(B,C)，过滤阶段才加载完整字段，减少数据传输量；
• 缓存复用：将validBCPairs缓存到内存/磁盘，避免多次关联时重复执行统计逻辑；
• 并行度调优：根据集群CPU/内存调整spark.sql.shuffle.partitions参数，优化shuffle性能；
• 存储格式：若原表是Parquet/Orc列式存储，读取效率远高于CSV，优先选用列式格式；
• 小样本验证：先拿1-2张测试表验证逻辑，再全量运行。

非Spark环境（如Hive）的实现思路 #

1. 创建临时视图关联所有表：

CREATE VIEW all_bc AS
SELECT B, C FROM tbl_1
UNION ALL
SELECT B, C FROM tbl_2
...
SELECT B, C FROM tbl_100;

2. 统计有效(B,C)组合：

CREATE TABLE bc_valid AS
SELECT B, C, COUNT(*) AS cnt
FROM all_bc
GROUP BY B, C
HAVING cnt >= 10;

3. 逐表过滤导出：

-- 示例处理tbl_1
INSERT OVERWRITE TABLE filtered_tbl_1
SELECT A, B, C FROM tbl_1
JOIN bc_valid ON tbl_1.B = bc_valid.B AND tbl_1.C = bc_valid.C;

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