使用 Sparklyr 在 R 中创建惰性 Spark 读写 Parquet 作业答案

【问题标题】：Create Lazy Spark Read-Write Parquet Job in R using Sparklyr使用 Sparklyr 在 R 中创建惰性 Spark 读写 Parquet 作业
【发布时间】：2018-11-23 23:13:55
【问题描述】：

我想创建一个从 sql 源读取的 spark 作业（使用“spark_read_jdbc”），然后将结果写入 parquet 文件（“spark_write_parquet”）。

对于sql语句中的小增量，该操作需要多次执行。我希望这份工作可以懒惰地创建，以便多个执行者可以接手这份工作。这是一些示例玩具代码：

sql = "SELECT * FROM TBL_%s"
for(i in seq(1,10)){
  sql_to_read = sprintf(sql, i)
  optionSet$dbtable = paste0("(", sql_to_read ,") foo")
  TEMP = spark_read_jdbc(sc, "TEMP", options = optionSet, memory = FALSE)
  TEMP = TEMP %>% mutate(id := i)
  TEMP %>% spark_write_parquet(., path = "/TEMP.parquet", mode = "append", partition_by = id)
}

问题是写操作不能强制执行串行操作。有没有办法编写此代码，以便为完整操作创建一个 spark 作业，然后当我启动一个“收集”语句时，多个执行器将执行该操作？

【问题讨论】：

在 scala 中使用线程解决了类似的问题：hadoopist.wordpress.com/2017/02/03/…
为什么不使用外部编排？例如，这看起来可以通过 Airflow 完成，只需传递参数数组。话虽这么说，为了能够在您需要线程时提交多个作业，而 R 没有本机多线程。将它委托给 JVM 应该是完全可能的，但老实说看起来很麻烦。您还可以在单独的进程中为每个作业建立新的连接。
最好将所有内容保存在一个环境中。 spark 的目的是分配处理（跨本地执行器或机器集群），因此设计为并行架构。奖励将是创建惰性 spark_write_parquet 作业，然后在设计完整作业后将其交给 spark 执行。
这里没有冲突。像这样的简单 Spark 迁移可以通过单个操作符轻松处理，几乎不需要代码。无论如何，我想这里的信息是 - 你可以做到，但实现这样的行为将需要在单线程 R 环境之外委派至少部分处理。以正确的方式做到这一点，尤其是处理部分故障，需要大量的工作。

标签： r apache-spark dplyr sparkr sparklyr

【解决方案1】：

您始终可以使用代码结构来完全避免该问题。因为所有写入都使用相同的输出表，您可以将输入定义为单个联合：

xs <- 1:10
query <- glue::glue("SELECT {xs} AS _id, * FROM TBL_{xs}") %>% 
  glue::collapse(" UNION ALL \n")

然后

optionSet$dbtable <- glue::glue("({query}) tmp")

spark_read_jdbc(sc, "TEMP", options = optionSet, memory = FALSE) %>%
  spark_write_parquet(., path = "/TEMP.parquet", mode = "append", partition_by = id)

您还可以使用id 作为partitionColumn 和min(xs) / max(xs) 作为lowerBound / upperBound 到parallelize reads。

【讨论】：

不是我所期望的，但值得奖励。