在 Spark 中执行以“WITH”关键字开头的 SQL 语句

Posted 2023-04-17

【中文标题】在 Spark 中执行以“WITH”关键字开头的 SQL 语句

【英文标题】：Execute SQL statement starting with "WITH" keyword in Spark

【发布时间】：2021-08-09 07:41:56

【问题描述】：

以下查询在 SQL Server 中运行良好：

with tt AS
(
select schema_name(fk_tab.schema_id) + '.' + fk_tab.name as foreign_table,
'>-' as rel,
schema_name(pk_tab.schema_id) + '.' + pk_tab.name as primary_table,
fk_cols.constraint_column_id as no,
fk_col.name as fk_column_name,
' = ' as [join],
pk_col.name as pk_column_name,
fk.name as fk_constraint_name
from sys.foreign_keys fk
inner join sys.tables fk_tab
on fk_tab.object_id = fk.parent_object_id
inner join sys.tables pk_tab
on pk_tab.object_id = fk.referenced_object_id
inner join sys.foreign_key_columns fk_cols
on fk_cols.constraint_object_id = fk.object_id
inner join sys.columns fk_col
on fk_col.column_id = fk_cols.parent_column_id
and fk_col.object_id = fk_tab.object_id
inner join sys.columns pk_col
on pk_col.column_id = fk_cols.referenced_column_id
and pk_col.object_id = pk_tab.object_id
)
 select tt.foreign_table, tt.primary_table, tt.fk_constraint_name
 ,(
SELECT SUBSTRING(
(
SELECT ',' + t.fk_column_name
FROM tt as t
WHERE tt.fk_constraint_name = t.fk_constraint_name FOR XML PATH('')), 2, 200000)
) AS fk_constraint_name
,(
SELECT SUBSTRING(
(
SELECT ',' + t.pk_column_name
FROM tt as t
WHERE tt.fk_constraint_name = t.fk_constraint_name FOR XML PATH('')), 2, 200000)
) AS pk_column_name
from tt
group by tt.foreign_table, tt.primary_table, tt.fk_constraint_name

它返回一个包含我数据库中所有外键的表。

我现在正尝试在 Databricks 上的 Spark 脚本中执行它。 这就是我所做的。

// ch contains my query
val df=spark.sql(ch)

但我收到此错误：

ParseException: 
no viable alternative at input 'with tt AS\n(\nselect schema_name(fk_tab.schema_id) + '.' + fk_tab.name as foreign_table,\n'>-' as rel,\nschema_name(pk_tab.schema_id) + '.' + pk_tab.name as primary_table,\nfk_cols.constraint_column_id as no,\nfk_col.name as fk_column_name,\n' = ' as

难道不能用 spark sql 执行这样的查询吗？ 如果没有，请问我该怎么做？

【问题讨论】：

分而治之：从最低限度的工作 CTE 开始，然后分阶段开始构建它，直到遇到错误。

【参考方案1】：

好吧，我刚刚想通了，对于这样的请求，应该使用 createStatement：

import java.sql._
val connection = DriverManager.getConnection(jdbcUrl)
val stmt1 = connection.createStatement 
val rs: ResultSet = stmt1.executeQuery(ch)

while (rs.next) 
  println(rs.getString("col1"))
  // ...