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

Transfer data from database to Spark using sparklyr

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

我可以使用基于 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_*() functions for import, but nothing database related. sdf_import() 看起来有可能,但不清楚如何在这种情况下使用它。

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)
    

    您可以使用 environment:

    一次传递多个 options
    invoke("options", as.environment(list(
      user="scott", password="tiger", url="jdbc:..."
    )))
    
  • 加载临时视图 dplyr:

    dplyr::tbl(sc, name)
    
  • 请务必阅读更多 JDBC 选项,重点关注 partitionColumn*BoundnumPartitions

  • 有关更多详细信息,请参见示例 and How to improve performance for slow Spark jobs using DataFrame and JDBC connection?