当您使用sparksql时，您可以使用sql解析器&它将为您完成这项工作。

def getTables(query: String): Seq[String] = {
  val logicalPlan = spark.sessionState.sqlParser.parsePlan(query)
  import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
  logicalPlan.collect { case r: UnresolvedRelation => r.tableName }
}

val query = "select * from table_1 as a left join table_2 as b on 
a.id=b.id"
scala> getTables(query).foreach(println)
table_1
table_2

您可以将“gettables”注册为udf并在查询中使用

您可以使用spark中的另一个sql函数collect\u listhttps://docs.databricks.com/spark/latest/spark-sql/language-manual/functions.html#collect_list. 你可以再找一个样品https://mungingdata.com/apache-spark/arraytype-columns/
基本上，应用到代码中

val df = spark.sql("select 1 id, 'select * from table_a, table_b' query" )
val df1 = spark.sql("select 2 id, 'select * from table_c join table_d' query" )

val df3 = df.union(df1)

df3.createOrReplaceTempView("tabla")

   spark.sql("""
select id, collect_list(tables) from (
        select id, explode(split(query, ' ')) as tables
        from tabla)
where tables like 'table%' group by id""").show

输出将是

+---+--------------------+
| id|collect_list(tables)|
+---+--------------------+
|  1| [table_a,, table_b]|
|  2|  [table_c, table_d]|
+---+--------------------+

希望这有帮助

如果你在 spark>=2.4 然后你可以移除 exploding 以及 collecting the same 在数组上使用高阶函数而不使用任何子查询的操作-

加载测试数据

val data =
      """
        |id  | query
        |1   | select * from table_a, table_b
        |2   | select * from table_c join table_d on table_c.id=table_d.id
      """.stripMargin
    val stringDS = data.split(System.lineSeparator())
      .map(_.split("\\|").map(_.replaceAll("""^[ \t]+|[ \t]+$""", "")).mkString(";"))
      .toSeq.toDS()
    val df = spark.read
      .option("sep", ";")
      .option("inferSchema", "true")
      .option("header", "true")
      .option("nullValue", "null")
      .csv(stringDS)
    df.printSchema()
    df.show(false)

    /**
      * root
      * |-- id: integer (nullable = true)
      * |-- query: string (nullable = true)
      *
      * +---+-----------------------------------------------------------+
      * |id |query                                                      |
      * +---+-----------------------------------------------------------+
      * |1  |select * from table_a, table_b                             |
      * |2  |select * from table_c join table_d on table_c.id=table_d.id|
      * +---+-----------------------------------------------------------+
      */

从查询中提取表

// spark >= 2.4.0
    df.createOrReplaceTempView("queries")
    spark.sql(
      """
        |select id,
        |   array_distinct(
        |   FILTER(
        |     split(query, '\\.|=|\\s+|,'), x -> x rlike 'table_a|table_b|table_c|table_d'
        |     )
        |    )as tables
        |FROM
        |   queries
      """.stripMargin)
      .show(false)

    /**
      * +---+------------------+
      * |id |tables            |
      * +---+------------------+
      * |1  |[table_a, table_b]|
      * |2  |[table_c, table_d]|
      * +---+------------------+
      */

如果只需要检查几个值，则可以使用 contains 函数而不是regexp:

val names = Seq("table_a","table_b","table_c","table_d")
def c(col: Column) = names.map(n => when(col.contains(n),n).otherwise(""))
df.select('id,array_remove(array(c('query):_*),"").as("result")).show(false)

但使用regexp时，它将如下所示（spark sql api）：

df.select('id,explode(split('query," ")))
.select('id,regexp_extract('col,"table_a|table_b|table_c|table_d",0).as("rx"))
.filter('rx=!="")
.groupBy('id)
.agg(collect_list('rx))

可以转换为以下sql查询：

select id, collect_list(rx) from
    (select id, regexp_extract(col,'table_a|table_b|table_c|table_d',0) as rx from
        (select id, explode(split(query,' ')) as col from df) q1
    ) q2
where rx != '' group by id

因此，输出将是：

+---+------------------+
| id|  collect_list(rx)|
+---+------------------+
|  1|[table_a, table_b]|
|  2|[table_c, table_d]|
+---+------------------+