使用 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 这样一次插入多行数据的东西。但并没有超过红移的行/大小限制。这对于简单的数据结构来说很容易，但是一个可以处理整数、字符、日期等的通用函数将是理想的，因为我不是只用一个表来做这件事的。因此，我打开了sqlSave 的源代码并开始使用我自己的函数来构建多行插入，它将数据分块为 1000 个行块，并为每个块构建和执行查询。

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

有什么建议吗？

更新 1

感谢一些提示，我调查了RODBC::sqlSave 中的fast=TRUE 开关。文档听起来像是我所追求的：

快：合乎逻辑。如果为 false，则一次写入一行数据。如果为 true，则使用参数化的 INSERT INTO 或 UPDATE 查询在一次操作中写入所有数据。

所以我想我应该测试一下。所以我创建了一个包含 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)
}

Answer 2

我在尝试解决为什么从 Pandas 写出约 5000 行数据框（通过 reticulate 导入到 R）需要很长时间并想发布更新时发现了这个答案。

从 0.24.0 版（2019 年 1 月）开始，pd.DataFrame.to_sql 中有一个 method 参数，当您设置 method = 'multi' 时，它似乎实际上进行了合理的分块。比较两种将数据帧写入 Redshift 的方法，我发现速度大约提高了 50 倍。

# Called from R, takes ~500 seconds for a 5000 row data frame
write_result <- pd_df$to_sql( name=tablename
    , con=conn
    , index = FALSE
    , if_exists = if_exists
    , schema='my_schema'
    , chunksize=1000L)

# Takes ~10 seconds for the same 5000 row data frame.
write_result <- pd_df$to_sql( name=tablename
    , con=conn
    , index = FALSE
    , if_exists = if_exists
    , schema='my_schema'
    , chunksize=1000L
    , method='multi') # this is the new argument

Answer 3

建议使用非 S3 选项进行批量加载：

1. 从 Amazon EMR 复制
2. 从远程主机 (SSH) 复制
3. 从 Amazon DynamoDB 复制

https://docs.aws.amazon.com/redshift/latest/dg/copy-parameters-data-source.html