将 R 代码推送到 postgresql 数据库中的 PL/R 代码

Question

我正在尝试将使用 Rpostgresql 的成功 R 代码转换为 PL/R 代码，以避免将数据推入和拉出 postgreql 数据库。

代码是 data.table 上的 dcast：

#libs
library(RPostgreSQL);
library(data.table);

# connect
drv <- dbDriver("PostgreSQL");
con <- dbConnect(drv, dbname="postgres", user="postgres");

# load
cli_ranges <- dbGetQuery(con, "SELECT custid, prod_ranges, is_cli from cli_ranges;")

# DT 
setDT(cli_ranges )
setkeyv(cli_ranges , c("prod_ranges"))

# pivot
cli_ranges.pivoted <- dcast(cli_ranges, custid ~ paste0("is_cli_", prod_ranges), fun=sum, value.var = "is_cli")

# send back to DB
dbWriteTable(con, "cli_ranges_pivoted", cli_ranges.pivoted, row.names=F)

R 中的代码运行良好且快速。

我现在正在尝试将代码推送到 PL/R 函数中，

CREATE OR REPLACE FUNCTION public.pivot()
  RETURNS void AS
$BODY$

[copy/paste R code]

$BODY$
  LANGUAGE plr;

...但是 R 代码的最后一行 (dbWriteTable) 抛出：

ERROR:  R interpreter expression evaluation error
DETAIL:  Error in (function (classes, fdef, mtable)  : 
  unable to find an inherited method for function 'dbWriteTable' for signature '"logical", "character", "data.frame"'
CONTEXT:  In PL/R function pivot

将 data.table 更改为数据框 (as.data.frame(cli_ranges.pivoted)) 也不起作用。

一个技巧可能是返回 data.table/frame 以执行CREATE TABLE cli_ranges_pivoted AS SELECT pivot();，但我真的不知道如何将 data.frame 作为输出推送...

cli_ranges 表：

custid  prod_ranges is_cli
1       A           1
1       B           1
1       C           0
2       A           1
2       B           0
2       C           1
3       A           0
3       B           1
3       C           0
4       A           1
...     ...         ...

dcast（即旋转）后数据帧如下：

custid  prod_ranges_A   prod_ranges_B   prod_ranges_C
1       1               1               0
2       1               0               1
3       0               1               0
4       1               ...
...

prod_ranges 中不同值的数量经常变化，因此我可以提前定义旋转后的列数。

环境：Postgresql 9.5、R 3.3、PL/R 08.03.00.16、Win 10 64 位

Answer 1

您可以从 http://gpdb.docs.pivotal.io/4330/ref_guide/pl_r.html 示例 2 和 3 开始。

或者，您可以尝试像 Lukas Eklund 和 Erwind Brandstetter 在这篇文章中所做的那样修改 unnest 函数（我正在使用 Lukas 的解决方案）： Unnest array by one level.

CREATE OR REPLACE FUNCTION unnest_multidim(anyarray)
RETURNS SETOF anyarray AS
$BODY$
  SELECT array_agg($1[series2.i][series2.x]) FROM
    (SELECT generate_series(array_lower($1,2),array_upper($1,2)) as x, series1.i
     FROM 
     (SELECT generate_series(array_lower($1,1),array_upper($1,1)) as i) series1 
    ) series2
GROUP BY series2.i
$BODY$
LANGUAGE sql IMMUTABLE;

然后您可以尝试返回一个数组，然后执行以下操作：

CREATE OR REPLACE FUNCTION r_norm(n integer, mean float8, std_dev float8)
 RETURNS float8[] 
AS $$
 x<-rnorm(n,mean,std_dev); 
 y<-rnorm(n,mean,std_dev);
 final<-cbind(as.data.frame(x), as.data.frame(y));
return(final) 

$$ LANGUAGE 'plr';

CREATE TABLE test_norm_var AS SELECT R_output[1] as col1, R_output[2] as col2 FROM unnest_multidim(r_norm(10,0,1)) R_output;

SELECT col1 FROM test_norm_var;

编辑

我无法让 dbWriteTable 像预期的 PL/R 函数那样工作... 但是，你也可以试试这个方法

CREATE OR REPLACE FUNCTION pivot() 
  RETURNS VOID as $$

library(RPostgreSQL);
library(data.table);

drv <- dbDriver("PostgreSQL");
con <- dbConnect(drv, dbname ="postgres");

fields <- list(custid = "numeric",prod_ranges = "varchar(128)", is_cli = "numeric")

custid <- c(1,1,1,2,2,2)
prod_ranges <- c("A","B","C","A","B","C")
is_cli <- c(1,1,0,1,0,1)

cli_ranges <- data.frame(custid,prod_ranges,is_cli, stringsAsFactors = default.stringsAsFactors())

setDT(cli_ranges )
setkeyv(cli_ranges , c("prod_ranges"))

cli_ranges.pivoted <- as.data.frame(dcast(cli_ranges, custid ~ paste0("is_cli_", prod_ranges), fun=sum, value.var = "is_cli"))

create_query <- paste0("CREATE TABLE cli_ranges (",paste0(colnames(cli_ranges.pivoted), collapse = " numeric, "),
" numeric) DISTRIBUTED BY (",colnames(cli_ranges)[1],")")

dbGetQuery(con, create_query);

values_string <- "("
for ( i in 1:dim(cli_ranges.pivoted)[1]){
 for ( j in 1:dim(cli_ranges.pivoted)[2] ){
    if ( j != dim(cli_ranges.pivoted)[2]) { 
        values_string <- paste0(values_string,cli_ranges.pivoted[i,j],",")
    } else { 
        values_string <- paste0(values_string,cli_ranges.pivoted[i,j])
    }
  } 
    if ( i != dim(cli_ranges.pivoted)[1] ){
        values_string <- paste0(values_string,"),(")
    } else {
        values_string <- paste0(values_string,")")
    }
}

insert_query <- paste0("INSERT INTO cli_ranges (",paste0(colnames(cli_ranges.pivoted), collapse = ", "),
") VALUES ", values_string )

dbGetQuery(con, insert_query);

$$ LANGUAGE plr;