在 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 的变量引发查询只是为了在多百行 f 字符串中插入 {var1}、{var2} 等。 我知道如何做到这一点，但它会很混乱、困难、难以阅读、迁移速度较慢并且维护起来更糟，如果可能的话，我想避免这种情况。

Answer 1

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

SET myVar FLOAT = NULL;
...

希望对您有所帮助:)

Answer 2

使用的 SET 命令用于 spark.conf get/set，而不是 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 的名称上使用 "dot"（或其他特殊字符），否则 SQL 单元格将期望您提供值到运行时间的 $variable（对我来说这看起来像是一个错误，我相信用 {} 舍入应该足够了）

Answer 3

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

首先我创建了一个示例 table:

%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

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

Assign a variable a dynamic value in SQL in Databricks / Spark

apache-spark

apache-spark-sql

pyspark-sql

databricks