在 Databricks / Spark 中的 SQL 中为变量分配动态值

Question

我觉得我必须在这里遗漏一些明显的东西，但我似乎无法在 Spark SQL 中动态设置变量值。

假设我有两个表，tableSrc 和 tableBuilder，我正在创建 tableDest。

我一直在尝试变体

SET myVar FLOAT = NULL

SELECT
    myVar = avg(myCol)
FROM tableSrc;

CREATE TABLE tableDest(
    refKey INT,
    derivedValue FLOAT
);


INSERT INTO tableDest
    SELECT
        refKey,
        neededValue * myVar AS `derivedValue`
    FROM tableBuilder

在 T-SQL 中执行此操作是微不足道的，这对 Microsoft 来说是一个令人惊讶的胜利 (DECLARE...SELECT)。然而，Spark 会抛出

Error in SQL statement: ParseException: mismatched input 'SELECT' expecting <EOF>(line 53, pos 0)

但我似乎无法将派生值分配给变量以供重用。我尝试了一些变体，但最接近的是将变量分配给 select 语句的字符串。

请注意，这是从 T-SQL 中的一个功能齐全的脚本改编而来的，所以我不会拆分十几个 SQL 变量来计算所有这些变量，使用 Python spark 查询来插入 @ 987654330@、{var2} 等在数百行 f 字符串中。 我知道如何做到这一点，但它会变得混乱、困难、难以阅读、迁移速度较慢、维护起来也更糟，因此我希望尽可能避免这种情况。

Answer 1

您在变量赋值的末尾缺少一个分号。

SET myVar FLOAT = NULL;
...

希望对你有帮助:)

Answer 2

SET 命令用于 spark.conf 获取/设置，而不是用于 SQL 查询的变量

对于 SQL 查询，您应该使用小部件：

https://docs.databricks.com/notebooks/widgets.html

但是，有一种方法可以在 SQL 上使用 spark.conf 参数：

%python spark.conf.set('personal.foo','bar')

那么你可以使用：

$sql select * from table where column = '${personal.foo}';

技巧部分是您必须在 spark.conf 的名称上使用“点”（或其他特殊字符），否则 SQL 单元将期望您在运行时为 $variable 提供值（看起来对我来说就像一个错误，我相信用 {} 进行四舍五入就足够了）

Answer 3

我在这个问题上盘旋了很长时间。最后，我找到了使用@Ronieri Marques 解决方案和一些 pyspark 函数的解决方法。我将尝试在下面提供完整的工作代码：

首先我创建一个示例表：

%sql
create table if not exists calendar
as 
select '2021-01-01' as date
union
select '2021-01-02' as date
union
select '2021-01-03' as date

%sql 
-- just to show the max and min dates
select max(date), min(date) from calendar

结合 sqlContext + toJSON 可以为变量动态赋值，在这种情况下我使用查询：

%python
result = sqlContext.sql("select max(date), min(date) from calendar").toJSON()
spark.conf.set('date.end'    , result.first()[14:24])
spark.conf.set('date.start'  , result.first()[39:49])

终于可以在 SQL 查询中使用变量了：

%sql 
select * from calendar where date > '${date.start}' and date < '${date.end}'

请注意，子字符串 result.first()[14:24] 和 result.first()[39:49] 是必需的，因为 result 的值。 first() 是 {"max(date)":"2021-01-03","min(date)":"2021-01-01"} 所以我们需要“定制”最终结果，只选择我们需要的值。

可能代码可以完善，但现在它是我设法实现的唯一可行的解​​决方案。

我希望这个解决方案对某人有用。

Answer 4

Databricks 刚刚发布了SQL user defined functions，它可以处理类似的问题而没有性能损失，对于您的示例，它看起来像：

CREATE TEMP FUNCTION myVar()
RETURNS FLOAT
LANGUAGE SQL
RETURN 
SELECT
    avg(myCol)
FROM tableSrc;

然后供使用：

SELECT
      refKey,
      neededValue * myVar() AS `derivedValue`
FROM tableBuilder