SparkR 和 Sparklyr 是兩個基於 Spark 的R語言接口，通過簡單的語法深度集成到R語言生態中。SparkR 由 Spark 社區維護，通過源碼級別更新 SparkR 的最新功能，最初從2016年夏天的1.5版本開始支持，從使用上非常像Spark Native。Sparklyr 由 RStudio 社區維護，通過深度集成 RStudio 的方式，提供更易於擴展和使用的方法，更強調統計特性與機器學習，實現本地與分布式代碼的一致性，通常會比 SparkR 延遲1-2個版本，從使用上看接近於dplyr。
由於 SparkR 與 Sparklyr 都是 Spark API 的封裝，故二者在計算性能上沒有顯著差異。
兩者文檔相對來說 Sparklyr 更加豐富一些，其中包含了業界/學界大量案例，但以中文版較少。SparkR 由第三方提供了中文版文檔。
SparkR 文檔：http://spark.apachecn.org/docs/cn/2.3.0/structured-streaming-programming-guide.html
Sparklyr 文檔：https://spark.rstudio.com
SparkR: 從官網下載。
Sparklyr:
sparklyr::spark_install(version = "2.3.0", hadoop_version = "2.7")
不依賴於Spark版本，spark 2.X 完美兼容1.X。
Spark 環境配置需要注意的問題：
下載和 Hadoop 對應版本號的發行版，具體可以通過 sparklyr::spark_available_versions() 查詢可用的spark版本
JAVA_HOME/SPARK_HOME/HADOOP_HOME 是必須要指定的環境變量，建議使用 JDK8/spark2.x/hadoop2.7
yarn-client/yarn-cluster 模式需要設置環境變量
Sys.setenv("HADOOP_CONF_DIR"="/etc/hadoop/conf")
連接 Hive 需要提供 Hive 連結配置, 在 spark-connection 初始化時指定對應 hive-site.xml 文件
由於不同發行版本的 Hadoop/Yarn 集群略有差異，環境 Setup 問題可以留言討論。
SparkR:
Sys.setenv("SPARKR_SUBMIT_ARGS"="--master yarn-client sparkr-shell")
sc <- SparkR::sparkR.session(enableHiveSupport = T,
sparkHome = "/data/FinanceR/Spark")
Sparklyr:
sc <- sparklyr::spark_connect(master = "yarn-client",
spark_home = "/data/FinanceR/Spark",
version = "2.2.0",
config = sparklyr::spark_config())
以寫Parquet文件為例,同理你可以用 SparkR::write.*()/sparklyr::spark_write_*() 等寫入其他格式文件到HDFS上,比如 csv/text。
什麼是 Parquet 文件？ Parquet 是一種高性能列式存儲文件格式，比CSV文件強在內建索引，可以快速查詢數據，目前普遍應用在模型訓練過程。
SparkR：
df <- SparkR::as.DataFrame(faithful)
SparkR::write.parquet(df,path= "/user/FinanceR",mode="overwrite",partition_by = "dt")
Sparklyr:
df <- sparklyr::copy_to(sc,faithful,"df")
sparklyr::spark_write_parquet(df,path="/user/FinanceR",mode="overwrite",partition_by = "dt")
dplyr 集成 spark/mysql 需要用到遠程處理模式。它要求先定義數據源表，再通過一系列dplyr操作惰性求值，直到執行 head() 或者 collect() 等觸發函數，才會執行計算過程，並將數據返回。如此設計是因為大數據集如果立即處理是無法優化數據處理流程的，通過惰性求值的方式，系統會在遠程機器上自動優化數據處理流程。
以統計計數為例：
從 db.financer_tbl 表中給 b 列 +2 後賦值為 a，過濾出 a > 2 條件下，每個 key 對應出現的次數並按照升序排序，最後去除統計缺失值。
SparkR:
library(SparkR)
library(magrittr)
remote_df = SparkR::sql("select * from db.financer_tbl limit 10")
remote_df %>%
mutate(a = df$b + 2) %>%
filter("a > 2")%>%
group_by("key")%>%
count()%>%
withColumn("count","cnt")%>%
orderBy("cnt",decrease = F)%>%
dropna() ->
pipeline
pipeline %>% persist("MEM_AND_DISK")
pipeline %>% head()
Sparklyr:
library(sparklyr)
library(dplyr)
remote_df = dplyr::tbl(sc,from = "db.financer_tbl")
remote_df %>%
mutate(a = b+2) %>%
filter(a > 2)%>%
group_by(key)%>%
summarize(count = n())%>%
select(cnt = count)%>%
order_by(cnt)%>%
arrange(desc(cnt))%>%
na.omit() ->
pipeline
pipeline %>% sdf_persist()
pipeline %>% head() %>% collect()
SparkR:
df <- SparkR::sql('SELECT * FROM financer_tbl WHERE dt = "20180318"')
Sparklyr:
由於 Sparklyr 通過 dplyr 接口操作，所以，所有數據操作幾乎和 MySQL 完全一樣，學習成本≈0。
df <- sc %>%
dplyr::tbl(dplyr::sql('SELECT * FROM financer_tbl WHERE dt = "20180318"'))
sc %>% DBI::dbGetQuery('SELECT * FROM financer_tbl WHERE dt = "20180318" limit 10') df %>% dbplyr::sql_render()
分發機制：
系統會將本地依賴文件壓縮打包上傳到HDFS路徑上，通過 Spark 動態分發到執行任務的機器上解壓縮。 執行任務的機器本地獨立的線程、內存中執行代碼，最後匯總計算結果到主要節點機器上實現R代碼的分發。
SparkR:
func <- function(x){x + runif(1) }
SparkR::gapplyCollect(x = df, func = func,group = "key")
Sparklyr:
func <- function(x){x + runif(1) }
sparklyr::spark_apply(x = df,packages=T,name = c("key","value"),func =func,group = "key")
SparkR 手動通過 spark.addFile 加載本地依賴，而 Sparklyr 打包本地 R 包只需通過 package = TRUE 參數即可，最大化減少學習成本。
什麼是流式計算? 流式計算是介於實時與離線計算之間的一種計算方式，以亞秒級準實時的方式小批量計算數據，廣泛應用在網際網路廣告、推薦等場景。
SparkR:
stream <- SparkR::read.stream(
source = "kafka",
"kafka.bootstrap.servers" = "a1.financer.com:9092,
a2.financer.com:9092",
"subscribe" = "binlog.financer.financer")
stream %>%
SparkR::selectExpr( "CAST(key AS STRING)", "CAST(value AS STRING)") %>%
SparkR::selectExpr("get_json_object(value,'$.data') as data") %>%
SparkR::selectExpr("get_json_object(data,'$.ORDERID') as orderid"
,"get_json_object(data,'$.USERID') as userid"
,"get_json_object(data,'$.TS') as ts"
) %>%
SparkR::withWatermark("ts", "5 minutes") %>%
SparkR::createOrReplaceTempView("financer")
"
select userid,window.start as ts,count(1) as cnt
from financer
group by userid, window(ts, '5 seconds')
" %>%
SparkR::sql() %>%
SparkR::write.stream("console",outputMode = "complete") ->
query
Sparklyr: 暫時不支持流式計算，功能開發中。
什麼是圖計算? 圖計算是以「圖論」為基礎的對現實世界的一種「圖」結構的抽象表達，以及在這種數據結構上的計算模式。 通常，在圖計算中，基本的數據結構表達就是：
G = （V，E，D）
V = vertex （頂點或者節點）
E = edge （邊）
D = data （權重）
SparkR: 不直接支持 Graph Minining。
Sparklyr: 通過拓展程序，graphframes 實現圖挖掘，比如Pagerank、LPA等。
下面是一個通過 graphframes 實現 Pagerank 的例子：
library(graphframes)
highschool_tbl <- copy_to(sc, ggraph::highschool, "highschool")
vertices_tbl <- sdf_bind_rows(
highschool_tbl %>% distinct(from) %>% transmute(id = from),
highschool_tbl %>% distinct(to) %>% transmute(id = to)
)
edges_tbl <- highschool_tbl %>% transmute(src = from, dst = to)
gf_graphframe(vertices_tbl, edges_tbl) %>%
gf_pagerank(reset_prob = 0.15, max_iter = 10L, source_id = "1")
SparkR 不直接支持 Deep Learnig。Sparklyr 通過拓展程序 Rsparkling 實現深度學習，比如 Anto-Encoder。
目前，SparkR 僅在實時計算上領先於 Sparklyr，在圖計算、機器學習、深度學習等領域已經被拉開差距，在大多數場景下，Sparklyr將是一個更好的選擇，在不久的將來，Sparklyr也將集成Streaming模塊，屆時將全面覆蓋SparkR功能。
相比於 pandas 和 pyspark，R 和 SparkR 的差異更小，並且如果你已經掌握了 dplyr 操作 mysql 的方法，學習 Sparklyr 將變得十分容易，因為他們共用同一套數據處理的語法，使用spark幾乎只有參數配置的學習成本， 更多 Sparklyr 教程可見 Sparklyr 官網 spark.rstudio.com 以及 Sparklyr 使用手冊: https://github.com/rstudio/cheatsheets/raw/master/translations/chinese/sparklyr-cheatsheet_zh_CN.pdf
https://eddjberry.netlify.com/post/2017-12-05-sparkr-vs-sparklyr/
https://github.com/rstudio/sparklyr/issues/502
https://databricks.com/session/r-and-spark-how-to-analyze-data-using-rstudios-sparklyr-and-h2os-rsparkling-packages
https://github.com/kevinykuo/sparklygraphs
http://docs.h2o.ai/h2o/latest-stable/h2o-docs/data-science/deep-learning.html
https://github.com/rstudio/graphframes
作者：朱俊輝
審稿：郎大為、黃俊文、朱雪寧
點擊查看原文連結：https://cosx.org/2018/05/sparkr-vs-sparklyr/ 或「閱讀原文」
統計之都：專業、人本、正直的中國統計學社區。
關注方式：掃描下圖二維碼。或查找公眾帳號，搜索 統計之都 或 CapStat 即可。
往期推送：進入統計之都會話窗口，點擊右上角小人圖標，查看歷史消息即可。