可行性结论 #

完全可以通过SparkSQL的Catalyst模块（含AstBuilder与ParsePlan能力）实现你需求中提取FROM子句表的步骤，不管是Hive、Teradata还是其他主流SQL方言的日志，只要SparkSQL Parser能兼容（或通过扩展方言适配），就能完成解析与表提取。

实现大纲 #

1. 初始化SparkSQL解析器 #

先获取Spark的SQL Parser实例，默认适配Spark方言，若要兼容Hive可直接启用Hive支持；Teradata则需扩展方言（见后续补充）：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
val spark = SparkSession.builder()
  .appName("SQLTableAnalyzer")
  .enableHiveSupport() // 启用Hive方言支持
  .getOrCreate()
val sqlParser: ParserInterface = spark.sessionState.sqlParser

2. 解析SQL生成LogicalPlan #

通过 parsePlan 将SQL字符串转为Catalyst的逻辑执行计划，这是解析的核心入口：

// 封装解析逻辑，处理语法错误等异常
def parseSql(sqlStr: String): Option[LogicalPlan] = {
  try {
    Some(sqlParser.parsePlan(sqlStr))
  } catch {
    case _: Exception => None // 跳过解析失败的SQL

3. 遍历LogicalPlan提取表名 #

递归遍历逻辑计划节点，提取所有涉及的表（处理JOIN、子查询、别名等场景）：

import org.apache.spark.sql.catalyst.plans.logical._
def extractTableNames(plan: LogicalPlan): Set[String] = plan match {
  // 直接匹配表扫描节点，提取完整表名（含库名如db.table）
  case scan: TableScan => Set(scan.tableName)
  // 子查询别名，递归处理内部节点
  case alias: SubqueryAlias => extractTableNames(alias.child)
  // JOIN场景，合并左右两边的表
  case join: Join => extractTableNames(join.left) ++ extractTableNames(join.right)
  // 投影、过滤等上层操作，递归处理子节点
  case project: Project => extractTableNames(project.child)
  case filter: Filter => extractTableNames(filter.child)
  // UNION场景，提取所有分支的表
  case union: Union => union.children.flatMap(extractTableNames).toSet
  // 其他节点类型，遍历所有子节点继续提取
  case _ => plan.children.flatMap(extractTableNames).toSet

4. 批量处理日志并统计频次 #

利用Spark分布式能力处理百万级日志，统计表的访问频次：

// 读取HDFS上的SQL日志（每行一条SQL）
val sqlLogsDF = spark.read.textFile("/hdfs/path/to/sql/logs")
// 解析每条SQL，提取表名并统计频次
val tableFreqResult = sqlLogsDF
  .flatMap(sql => parseSql(sql).flatMap(plan => extractTableNames(plan).toSeq))
  .groupBy("value")
  .count()
  .orderBy(org.apache.spark.sql.functions.desc("count"))
// 输出Top N结果
tableFreqResult.show(50)
// 保存结果到Hive表
tableFreqResult.write.mode("overwrite").saveAsTable("dw.most_accessed_tables")

补充：Teradata方言适配 #

如果遇到Teradata特有语法导致解析失败，可自定义方言扩展：

import org.apache.spark.sql.catalyst.parser.Dialect
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
class TeradataDialect extends Dialect {
  override def parse(sql: String): LogicalPlan = {
    // 参考HiveDialect实现，扩展SqlBaseParser支持Teradata语法
    // 实际项目中可复用现有开源适配或自行实现关键语法解析