Scala替代嵌套for循环 优化Postgres多表删除性能

你当前的实现性能差完全是交互逻辑设计不合理导致的，核心问题有三个：

• 两层循环逐表、逐ID发送单条DELETE语句，假设坏ID有1000条、目标表100张，就会产生10万次数据库网络往返，绝大部分时间都耗在网络IO上
• 每条DELETE单独触发事务提交、索引检索、行锁申请，数据库重复执行大量相同的前置逻辑，资源浪费严重
• 现有代码存在两处显性运行时错误：一是读取表名列表时错误选择了corrput_id字段，实际SQL查询返回的字段是table_name；二是坏ID列表变量名前后不一致（定义时叫badrecods，循环时用bad_records），直接运行会抛异常。

按改造成本从低到高，给两个可直接落地的方案：

方案1：批量IN删除（改造成本最低，性能提升10~100倍） #

把单表N次单ID删除，改成单表1次（或少量批次）IN条件匹配删除，把数据库请求次数从「表数量ID数量」降到「表数量批次数」，1000个ID以内的场景用这个方案足够。

修正前置逻辑 #

先修正之前的字段读取、变量名问题，同时做SQL转义避免语法错误和注入风险：

// 读取CSV坏ID
val csvDF = spark.read.format("csv")
  .option("header", "true")
  .option("delimiter", ",")
  .option("inferSchema", true)
  .option("escape", "\"")
  .option("multiline", "true")
  .option("quotes", "")
  .load(inputPath)
// 统一变量名，转义单引号防SQL语法错误
val badIds = csvDF.select("corrput_id").as[String].collect().toList
val escapedBadIds = badIds.map(id => s"'${id.replace("'", "''")}'")

// 读取PG表列表，修正字段读取错误
val query = s"(select table_name from information_schema.tables where table_schema = '${db}' and table_name not in ${excludetables}) temp "
val tablesdf = spark.read.jdbc(jdbcUrl, table = query, connectionProperties)
val tableList = tablesdf.select("table_name").as[String].collect().toList

批量删除逻辑 #

关闭JDBC自动提交，用批量执行减少事务开销，ID数量多的时候按每批1000个拆分，避免PG SQL长度限制：

val conn = dbconnection
conn.setAutoCommit(false)
try {
  tableList.foreach { tableName =>
    val stmt = conn.createStatement()
    try {
      // 每1000个ID拆成一批，避免单条SQL过长
      escapedBadIds.grouped(1000).foreach { idBatch =>
        val deleteSql = s"""
          DELETE FROM ${db}.${tableName}
          WHERE corrput_id IN (${idBatch.mkString(",")})
        """
        stmt.addBatch(deleteSql)
      }
      stmt.executeBatch()
      conn.commit()
      println(s"表 ${tableName} 清理完成")
    } finally {
      stmt.close()
    }
  }
} catch {
  case e: Exception =>
    conn.rollback()
    throw e
} finally {
  conn.setAutoCommit(true)
  conn.close()
}

方案2：临时表关联删除（适合坏ID>1万的大数量场景） #

如果坏ID数量过万，拼IN列表会出现SQL解析慢、长度超限的问题，这个场景下可以先把所有坏ID写入PG会话级临时表，再通过关联删除实现单表1次SQL完成清理，性能比IN方案更高，没有ID数量限制。

val conn = dbconnection
conn.setAutoCommit(false)
try {
  val stmt = conn.createStatement()
  // 创建会话临时表，会话断开自动删除，不占用永久存储
  stmt.execute("CREATE TEMP TABLE tmp_bad_ids (corrput_id VARCHAR PRIMARY KEY) ON COMMIT DROP")
  // 批量插入所有坏ID到临时表
  val insertPs = conn.prepareStatement("INSERT INTO tmp_bad_ids (corrput_id) VALUES (?)")
  badIds.foreach { id =>
    insertPs.setString(1, id)
    insertPs.addBatch()
  }
  insertPs.executeBatch()
  insertPs.close()
  conn.commit()

  // 逐表关联删除
  tableList.foreach { tableName =>
    val deleteSql = s"""
      DELETE FROM ${db}.${tableName} t
      USING tmp_bad_ids b
      WHERE t.corrput_id = b.corrput_id
    """
    stmt.execute(deleteSql)
    conn.commit()
    println(s"表 ${tableName} 清理完成")
  }
} catch {
  case e: Exception =>
    conn.rollback()
    throw e
} finally {
  conn.setAutoCommit(true)
  conn.close()
}

• 执行删除前先确认所有目标表的corrput_id字段存在索引，没有索引的话删除会触发全表扫描，大表下速度会极慢
• 清理操作尽量选业务低峰期执行，避免长事务持锁影响线上业务
• 不需要用Spark的分布式JDBC写入能力做这个操作，这类简单元数据驱动的删除场景，单连接批量执行的开销远低于Spark分布式任务调度的开销
• 禁止直接拼接未转义的ID值到SQL里，不仅会因为ID包含单引号触发语法错误，还存在SQL注入风险

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