sql - 使用 R 一次将多条记录插入 Redshift

Question

我需要每天多次将几千行推送到 Redshift 中。但出于管理原因，我不能使用 S3 中的批量插入。最快的方法是什么？

详情:

有 3 种方法(我可以看到)将行从 R 插入到 Amazon Redshift 中的表中:

1. 逐行插入查询。每行都作为自己的 INSERT VALUES 查询
插入
2. 多行插入查询:同1，但每次查询插入多条记录。大小受行数或 16MB 最大 SQL 查询大小限制的限制。
3. 从 AWS S3、Dynamo 或 EMR 批量插入。

上述每种方法都比之前的方法快一个数量级。我喜欢使用批量插入来创建或更新表，但我们的组织已做出安全和管理决定，不允许业务用户将数据批量加载或卸载到 S3。所以这让我无法使用包 redshiftTools上传文件。

我可以使用 RODBC::sqlSave 执行上面的第 1 条。这很慢，但最终完成了工作。

我宁愿拥有类似 sqlSave 的东西，它可以一次插入多行数据。但不要超过 Redshift 的行/大小限制。这对于简单的数据结构来说很容易，但是处理整数、字符、日期等的通用函数将是理想的，因为我不会只对一个表执行此操作。因此，我打开了 sqlSave 的源代码，并开始推出我自己的函数来构建多行插入，将数据分成 1000 行 block ，并为每个 block 构建和执行查询。

但是我在这里停下来问这是否已经完成了？有没有更好的办法？我有这种感觉，也许 R 的其他 SQL 包之一具有执行类似操作的功能。然而，当我搜索时，我发现的都是其他人 having the same sort of problem .

有什么建议吗？

更新 1

多亏了一些提示，我研究了 RODBC::sqlSave 中的 fast=TRUE 开关。文档听起来像是我所追求的:

fast: logical. If false, write data a row at a time. If true, use a parametrized INSERT INTO or UPDATE query to write all the data in one operation.

所以我想我应该对此进行测试。所以我创建了一个包含 10 条记录和 2 列的小数据框:

df <- data.frame(a=rnorm(10), b=sample(letters, 10, replace=TRUE),
                 stringsAsFactors = FALSE)    

然后我使用 benchmark 来计算 5 次复制的执行时间:

benchmark( sqlSave(dbhandle, df, tablename = 'jal_test1', append=TRUE, fast=TRUE), replications = 5)
#                                                                         test replications elapsed relative user.self sys.self user.child sys.child
# 1 sqlSave(dbhandle, df, tablename = "jal_test1", append = TRUE, fast = TRUE)            5  512.59        1      0.08     0.03         NA        NA

benchmark( sqlSave(dbhandle, df, tablename = 'jal_test1', append=TRUE, fast=FALSE), replications = 5)
#                                                                          test replications elapsed relative user.self sys.self user.child sys.child
# 1 sqlSave(dbhandle, df, tablename = "jal_test1", append = TRUE, fast = FALSE)            5  264.37        1      0.08     0.02         NA        NA
      

这有点难读，但总而言之:

• fast=TRUE 耗时 512 秒
• fast=FALSE 耗时 264 秒

有25条记录，时间上升到:

• fast=TRUE 耗时 1208 秒
• fast=FALSE 耗时 604 秒

这对我来说完全没有意义。

更新2

我尝试了 test=TRUE 开关，认为它会告诉我发生了什么，但我根本不知道它做了什么......但是转向 verbose=TRUE 帮助我意识到 fast=TRUE 并不像我想的那样。它似乎使用了替换，但没有做一个大的插入。它仍然执行 nrow(df) 的插入操作:

> df <- data.frame(a=rnorm(5), b=sample(letters, 5, replace=TRUE), stringsAsFactors = FALSE)
> sqlSave(dbhandle, df, tablename = 'jal_test1', append=TRUE, fast=FALSE, verbose=TRUE)
Query: INSERT INTO "jal_test1" ( "rownames", "a", "b" ) VALUES ( '1', -1.45261402, 'd' )
Query: INSERT INTO "jal_test1" ( "rownames", "a", "b" ) VALUES ( '2', -0.01642518, 'm' )
Query: INSERT INTO "jal_test1" ( "rownames", "a", "b" ) VALUES ( '3',  1.11767938, 'm' )
Query: INSERT INTO "jal_test1" ( "rownames", "a", "b" ) VALUES ( '4', -0.63480166, 'a' )
Query: INSERT INTO "jal_test1" ( "rownames", "a", "b" ) VALUES ( '5', -0.75538702, 'k' )
> sqlSave(dbhandle, df, tablename = 'jal_test1', append=TRUE, fast=TRUE, verbose=TRUE)
Query: INSERT INTO "jal_test1" ( "rownames", "a", "b" ) VALUES ( ?,?,? )
Binding: 'rownames' DataType -9, ColSize 255
Binding: 'a' DataType 6, ColSize 17
Binding: 'b' DataType -9, ColSize 255
Parameters:
no: 1: rownames 1/***/no: 2: a -1.45261/***/no: 3: b d/***/
no: 1: rownames 2/***/no: 2: a -0.0164252/***/no: 3: b m/***/
no: 1: rownames 3/***/no: 2: a 1.11768/***/no: 3: b m/***/
no: 1: rownames 4/***/no: 2: a -0.634802/***/no: 3: b a/***/
no: 1: rownames 5/***/no: 2: a -0.755387/***/no: 3: b k/***/

Answer 1

我最终无法找到可以在 R 中进行分块的 SQL 写入函数的实现。但我确实看到在 Python 中，sqlalchemy 包与 pandas 组合在一起可以轻松做到这一点。因此，我推出了 Reticulate 并将 Python 包装在一些 R 代码中，以创建一个函数来写入 redshift。看起来有点矫枉过正，但无需我重新实现任何东西就可以完成工作:

start_python <- function(){
  library(reticulate)
  use_condaenv( "r-reticulate")
  pd <- import('pandas')
  sa <- import('sqlalchemy')
}

# write a table to RDW sandbox
write_to_redshift <- function(df, tablename, if_exists = 'append'){
  pd_df <- r_to_py(df)
  eng = sa$create_engine('postgres://user:pwd@redshift_name:5439/db_name')
  conn = eng$connect()
  write_result <- pd_df$to_sql( name=tablename, con=conn,  index = FALSE, if_exists = if_exists, schema='my_schema', chunksize=10000L)
  conn$close()
  return(write_result)
}