如何在Databricks中的SQL查询中使用python变量？

Question

我正在尝试将 SQL 存储过程转换为 databricks 笔记本。在存储过程中要实现以下 2 条语句。这里表1和表2是databricks集群中的Delta Lake表。

我想在 SQL 查询中使用 python 变量代替 max_date 。怎么做？

%sql
DELETE FROM table1 WHERE Date = max_date;
INSERT INTO table1
SELECT * FROM table2 WHERE Date = max_date;

Answer 1

如果您要逐个单元地运行它，那么您可以使用 databricks 小部件，例如

第一个细胞

x=str(datetime.date.today())
dbutils.widgets.text("max_date",x)

第二个细胞

%sql
select getArgument("max_date") AS max_date

会给你

max_date
2022-06-04

但正如这里提到的，当使用 run all 时它不起作用，理想的方法是创建单独的基于语言的笔记本并使用 %run 传递变量

另一种方法是使用spark conf，如下所示

首先为conf设置一个值[注意-conf名称中应该有.(点)]

max_date2=str(datetime.date.today())
spark.conf.set("abc.max_dt2", max_date2)

接下来尝试选择如下所示的值

%sql
select "${abc.max_dt2}" as max_date

它应该给出与上面相同的值

Answer 2

假设您计算max_date或使用小部件来传递值，如下所示：

max_date = dbutils.widgets.get("max_date")

您可以spark.sql()在 Python 字符串中使用 SQL 查询，如下所示：

df = spark.sql(f"""
DELETE FROM table1 WHERE Date = '{max_date}';
INSERT INTO table1
SELECT * FROM table2 WHERE Date = '{max_date}';
""")

display(df)

使用 f 字符串格式根据需要制作 SQL 字符串，然后将其传递给内置的 Spark.sql() 执行器，会更容易且更具可读性。如果需要，spark.sql() 函数将返回一个包含 SQL 查询结果的 DataFrame。

更新：解决一个好评论！