如何使用 SparkR::read.jdbc() 或 sparklyr::spark_read_jdbc() 获取 SQL 查询而不是整个表的结果？

Question

我通常在本地使用 RODBC 来查询我的数据库。然而，我们公司最近迁移到 Azure Databricks，它本身并不支持 RODBC 或其他 odbc 连接，但确实支持我以前没有使用过的 jdbc 连接。

我已经阅读了 SparkR::read.jdbc() 和 sparklyr::spark_read_jdbc() 的文档，但是这些似乎从数据库中提取了整个表，而不仅仅是查询的结果，这不适合我因为我不必提取整个表，而是运行将多个表连接在一起的查询，但只返回每个表中非常小的数据子集。

我找不到使用 jdbc 连接器的方法：

(A) 运行引用同一数据库上的多个表的查询

和

(B) 将结果存储为 R 数据帧或可以很容易地转换为 R 数据帧的东西（例如 SparkR 或 sparklyr 数据帧）。

如果可能，该解决方案还只需要我为每个脚本/笔记本指定一次连接凭据，而不是每次连接到数据库以运行查询并将结果存储为数据框时。

例如是否有等效于以下内容的 jdbc：

my_server="myserver.database.windows.net"
my_db="mydatabase"
my_username="database_user"
my_pwd="abc123Ineedabetterpassword"


myconnection <- RODBC::odbcDriverConnect(paste0("DRIVER={SQL Server};
                                 server=",my_server,";
                                 database=",my_db,";
                                 uid=",my_username,";
                                 pwd=",my_pwd))

df <- RODBC::sqlQuery(myconnection, 
"SELECT a.var1, b.var2, SUM(c.var3) AS Total_Things, AVG(d.var4) AS Mean_Stuff
FROM table_A as a 
JOIN table_B as b on a.id = b.a_id
JOIN table_C as c on a.id = c.a_id
JOIN table_D as d on c.id = d.c_id
Where a.filter_var IN (1, 2, 3, 4)
AND d.filter_var LIKE '%potatoes%'
GROUP BY
a.var1, b.var2
")

df2 <- RODBC::sqlQuery(myconnection,
"SELECT x.var1, y.var2, z.var3
FROM table_x as x
LEFT JOIN table_y as y on x.id = y.x_id
LEFT JOIN table_z on as z on x.id = z.x_id
WHERE z.category like '%vegetable%'
AND y.category IN ('A', 'B', 'C')
“)

如何使用 SparkR 中的 jdbc 连接器或 Databricks 中内置的 sparklyr 执行与上述相同的结果（两个 R 数据帧 df 和 df2）？

我知道我可以使用 spark 连接器和一些 scala 代码 (https://docs.microsoft.com/en-us/azure/sql-database/sql-database-spark-connector) 将查询结果存储为 spark 数据框，将其转换为全局临时表，将全局临时表存储为 SparkR 数据框并折叠这是一个 R 数据框，但是这段代码很难阅读，需要我将笔记本中的一个单元格的语言更改为 scala （我不太了解），并且由于大而需要很长时间步数。因为我的 R 脚本通常以多个 SQL 查询开始——通常是针对多个不同的数据库——这种方法变得非常耗时并且使我的脚本几乎不可读。肯定有更直接的方法吗？

（我们使用 Databricks 主要通过 LogicApps 和 Azure 数据工厂实现自动化，偶尔用于增加 RAM，而不是用于并行处理；我们的数据（一旦提取）通常不足以需要并行化和我们的一些模型使用（例如 lme4::lmer()）不会从中受益。）

Answer 1

我最终解决了这个问题，并想在这里发布答案，以防其他人遇到问题。

您可以将 SparkR::read.jdbc() 与查询一起使用，但您必须将查询括在括号中并将结果别名为某事，否则您将得到一个模棱两可的语法错误。作为默认的 1433，“portnum”对我来说似乎可以正常工作，但如果您有不同类型的 SQL 数据库，您可能需要在 URL 中更改它。然后您可以在包含查询结果的 SparkDataFrame 上调用 SparkR::collect() 以将其转换为 R 数据帧：

例如

myconnection <- "jdbc:sqlserver://myserver.database.windows.net:1433;database=mydatabase;user=database_user;password=abc123Ineedabetterpassword"

df <- read.jdbc( myconnection, "(
SELECT a.var1, b.var2, SUM(c.var3) AS Total_Things, AVG(d.var4) AS Mean_Stuff
FROM table_A as a 
JOIN table_B as b on a.id = b.a_id
JOIN table_C as c on a.id = c.a_id
JOIN table_D as d on c.id = d.c_id
Where a.filter_var IN (1, 2, 3, 4)
AND d.filter_var LIKE '%potatoes%'
GROUP BY
a.var1, b.var2) as result" ) %>% 
SparkR::collect()