使用sparklyr将数据从数据库传输到Spark

Question

我在数据库中有一些数据,我想使用sparklyr在Spark中使用它.

我可以使用基于DBI的包将数据库中的数据导入到R中

dbconn <- dbConnect(<some connection args>)
data_in_r <- dbReadTable(dbconn, "a table") 

然后使用将数据从R复制到Spark

sconn <- spark_connect(<some connection args>)
data_ptr <- copy_to(sconn, data_in_r)

对于大数据集,复制两次很慢.

如何将数据直接从数据库复制到Spark？

sparklyr有几个spark_read_*()导入功能,但没有数据库相关. sdf_import()看起来像是一种可能性,但目前尚不清楚如何在这种情况下使用它.

Answer 1

Sparklyr> = 0.6.0

你可以用spark_read_jdbc.

Sparklyr <0.6.0

我希望有更优雅的解决方案,但这是使用低级API的最小示例:

• 确保Spark可以访问所需的JDBC驱动程序,例如通过添加其坐标spark.jars.packages.例如,使用PostgreSQL(调整当前版本),您可以添加:

  spark.jars.packages org.postgresql:postgresql:9.4.1212
  

  至 SPARK_HOME/conf/spark-defaults.conf

• 加载数据并注册为临时视图:

  name <- "foo"
  
  spark_session(sc) %>% 
    invoke("read") %>% 
    # JDBC URL and table name
    invoke("option", "url", "jdbc:postgresql://host/database") %>% 
    invoke("option", "dbtable", "table") %>% 
    # Add optional credentials
    invoke("option", "user", "scott") %>%
    invoke("option", "password", "tiger") %>% 
    # Driver class, here for PostgreSQL
    invoke("option", "driver", "org.postgresql.Driver") %>% 
    # Read and register as a temporary view
    invoke("format", "jdbc") %>% 
    invoke("load") %>% 
    # Spark 2.x, registerTempTable in 1.x
    invoke("createOrReplaceTempView", name)
  

  您可以options使用以下命令一次传递多个environment:

  invoke("options", as.environment(list(
    user="scott", password="tiger", url="jdbc:..."
  )))
  

• 加载临时视图dplyr:

  dplyr::tbl(sc, name)
  

• 请务必阅读关于进一步JDBC选项,重点是partitionColumn,*Bound和numPartitions.

• 有关其他详细信息,请参阅示例如何使用JDBC源在(Py)Spark中写入和读取数据？以及如何使用DataFrame和JDBC连接提高慢速Spark作业的性能？