For 循环从 R 中的函数设置周数

Posted 2023-02-14

【中文标题】For 循环从 R 中的函数设置周数

【英文标题】：For Loop For Set Weeks from Function in R

【发布时间】：2022-01-18 19:52:57

【问题描述】：

我正在使用 Databricks 从 Snowflake 中检索数据。目前在 R 中使用的函数按原样工作。但是，我的数据集非常大（100B 行），并且想以 for 循环的形式调用我的数据集以降低内存。如何根据 for 循环检索如下数据集？

## Call Libraries to retrieve dataset from Snowflake
library(rJava)
library(RJDBC)
library(DBI)

## Connect to Snowflake

jdbcDriver <- JDBC(driverClass="com.snowflake.client.jdbc.SnowflakeDriver",
                  classPath="dbfs:/FileStore/jars/ced63da8_4613_4687_9051_1ea61dbf3b76- 
snowflake_jdbc_3_12_12-30836.jar")

jdbcConnection <- dbConnect(jdbcDriver,
                           "jdbc:snowflake://xxx.us-east-1.snowflakecomputing.com/",
                           "xxxx",
                           dbutils.secrets.get(scope="xxx", key="xxx"))

下面是我希望实现 for 循环以检索每周数据而不是整个数据集并将结果彼此附加的地方。目标是根据这些结果在 Snowflake 中创建一个新表，但很好奇如何首先创建 for 循环。

## Run SQL Query to retrieve dataset from Snowflake in the form of an R function
sq <- function(query)
 dbGetQuery(jdbcConnection, knitr::knit_expand(text=query))



show1 <- function(a) 
  sq(
    paste(
"  
select
data_source,
media_id, 
sum(activity_count)
from", a, "
where START_DATE_EST between '2019-11-01' and '2021-11-14' "
)

)
activity_by_month<-show1("SCHEMA.TABLE_NAME")

如果我要运行它，这将需要几天的时间，并且由于数据量的原因它会超时。我想以 for 循环的格式运行此函数，在该循环中我按周收集数据并将结果附加在一起。以前有人这样做过吗？

理想的方法是这样运行：

start_date_vector <- c('2019-11-01', '2019-11-09', etc.)
end_date_vector <- c('2019-11-08', '2019-11-16', etc.)

for(i in start_date_vector) 
  for(j in end_date_vector) 
    show1

rbind(show1)

上面的 SQL 查询看起来像这样：

where START_DATE_EST between", start_date_vector," and",end_date_vector "

然后编写一个 SQL 查询，将结果合并在一起并保存一个表。

【问题讨论】：

看起来该模式正在将数据移出 Snowflake，在外部处理并将其保存回来。根据具体要求，可能值得考虑替代方法（即Snowpark 和Snowpark Overview），无需往返即可处理数据。

谢谢，是的，这就是概念！数据集太大而无法一次运行，因此我想将其拆分，以免 Databricks/Snowflake 内存不足。一些同事建议我做一个 for 循环。我也会研究这个解决方案！

【参考方案1】：

鉴于数据集很大，与通过 JDBC 检索数据以在 Spark (Databricks) 环境中使用相比，可能有几个更好、更快的选项，您可能希望在提取数据后并行进一步处理.

Snowflake Spark 连接器 该连接器可用，preloaded in Databricks 因此您应该能够在您的环境中使用它。它专门设计用于处理 Snowflake 和 Spark 集群之间的直接大规模数据传输，并在后台使用批量数据传输技术。它可以用于retrieve data from Snowflake into a Spark dataframe，然后可以在需要时持久化到存储中。

复制到分区的 PARQUET 文件 Snowflakes COPY INTO 命令支持分区文件卸载数据到 PARQUET 文件。您可以指定要用于对文件进行分区的列/表达式，类似于您在 R 中使用的嵌套 for 循环方法，例如 partition by ('date=' || to_varchar(START_DATE_EST), 'YYYY-MM') || '/week=' || to_varchar(week(START_DATE_EST))).

您可以将文件卸载到内部 Snowflake 阶段，然后将 GET 文件从那里卸载到本地 R 环境。如果您有权访问云存储（例如 AWS S3 存储桶），您可以通过 Snowflake External 阶段在那里卸载，该阶段也可以通过具有正确权限的 Spark 环境直接访问。 JDBC 和 ODBC 驱动程序支持 COPY INTO 和 GET 命令，因此您可以通过编程方式形成所需的 SQL 代码并从 R 中执行。