SparkR (R on Spark)

• 概述
• SparkDataFrame
  • 啟動: SparkSession
  • 從 RStudio 來啟動
  • 創建 SparkDataFrames
    • 從本地的 data frames 來創建 SparkDataFrames
    • 從 Data Sources（數據源）創建 SparkDataFrame
    • 從 Hive tables 來創建 SparkDataFrame
  • SparkDataFrame 操作
    • Selecting rows（行）, columns（列）
    • Grouping, Aggregation（分組, 聚合）
    • Operating on Columns（列上的操作）
    • 應用 User-Defined Function（UDF 用戶自定義函數）
      • Run a given function on a large dataset using dapply or dapplyCollect
        • dapply
        • dapplyCollect
      • Run a given function on a large dataset grouping by input column(s) and using gapply or gapplyCollect（在一個大的 dataset 上通過 input colums（輸入列）來進行 grouping（分組）並且使用 gapply or gapplyCollect
        來運行一個指定的函數）
        • gapply
        • gapplyCollect
      • 使用 spark.lapply 分發運行一個本地的 R 函數
        • spark.lapply
  • SparkR 中運行 SQL 查詢
• 機器學習
  • 算法
    • 分類
    • 回歸
    • 樹
    • 聚類
    • 協同過濾
    • 頻繁模式挖掘
    • 統計
  • 模型持久化
• R和Spark之間的數據類型映射
• Structured Streaming
• R 函數名沖突
• 遷移指南
  • SparkR 1.5.x 升級至 1.6.x
  • SparkR 1.6.x 升級至 2.0
  • 升級至 SparkR 2.1.0
  • 升級至 SparkR 2.2.0

概述

SparkR 是一個 R package, 它提供了一個輕量級的前端以從 R 中使用 Apache Spark. 在 Spark 2.2.0 中, SparkR 提供了一個分布式的 data frame, 它實現了像 selection, filtering, aggregation etc 一系列所支持的操作.（dplyr 與 R data frames 相似) ）, 除了可用於海量數據上之外. SparkR 還支持使用 MLlib 來進行分布式的 machine learning（機器學習）.

SparkDataFrame

SparkDataFrame 是一個分布式的, 將數據映射到有名稱的 colums（列）的集合. 在概念上 相當於關系數據庫中的 table 表或 R 中的 data frame，但在該引擎下有更多的優化. SparkDataFrames 可以從各種來源構造，例如: 結構化的數據文件，Hive 中的表，外部數據庫或現有的本地 R data frames.

All of the examples on this page use sample data included in R or the Spark distribution and can be run using the ./bin/sparkR shell.

啟動: SparkSession

SparkR 的入口點是 SparkSession, 它會連接您的 R 程序到 Spark 集群中. 您可以使用 sparkR.session 來創建 SparkSession, 並傳遞諸如應用程序名稱, 依賴的任何 spark 軟件包等選項, 等等. 此外，還可以通過 SparkSession 來與 SparkDataFrames 一起工作。 如果您正在使用 sparkR shell，那麽 SparkSession 應該已經被創建了，你不需要再調用 sparkR.session.

sparkR.session()

從 RStudio 來啟動

您可以從 RStudio 中來啟動 SparkR. 您可以從 RStudio, R shell, Rscript 或者 R IDEs 中連接你的 R 程序到 Spark 集群中去. 要開始, 確保已經在環境變量中設置好 SPARK_HOME (您可以檢測下 Sys.getenv), 加載 SparkR package, 並且像下面一樣調用 sparkR.session. 它將檢測 Spark 的安裝, 並且, 如果沒有發現, 它將自動的下載並且緩存起來. 當然，您也可以手動的運行 install.spark.

為了調用 sparkR.session, 您也可以指定某些 Spark driver 的屬性. 通常哪些 應用程序屬性 和 運行時環境 不能以編程的方式來設置, 這是因為 driver 的 JVM 進程早就已經啟動了, 在這種情況下 SparkR 會幫你做好準備. 要設置它們, 可以像在 sparkConfig 參數中的其它屬性一樣傳遞它們到 sparkR.session() 中去.

if (nchar(Sys.getenv("SPARK_HOME")) < 1) {
  Sys.setenv(SPARK_HOME = "/home/spark")
}
library(SparkR, lib.loc = c(file.path(Sys.getenv("SPARK_HOME"), "R", "lib")))
sparkR.session(master = "local[*]", sparkConfig = list(spark.driver.memory = "2g"))

下面的 Spark driver 屬性可以 從 RStudio 的 sparkR.session 的 sparkConfig 中進行設置:

創建 SparkDataFrames

有了一個 SparkSession 之後, 可以從一個本地的 R data frame, Hive 表, 或者其它的 data sources 中來創建 SparkDataFrame 應用程序.

從本地的 data frames 來創建 SparkDataFrames

要創建一個 data frame 最簡單的方式是去轉換一個本地的 R data frame 成為一個 SparkDataFrame. 我們明確的使用 as.DataFrame 或 createDataFrame 並且經過本地的 R data frame 中以創建一個 SparkDataFrame. 例如, 下面的例子基於 R 中已有的 faithful 來創建一個 SparkDataFrame.

df <- as.DataFrame(faithful)

# 展示第一個 SparkDataFrame 的內容
head(df)
##  eruptions waiting
##1     3.600      79
##2     1.800      54
##3     3.333      74

從 Data Sources（數據源）創建 SparkDataFrame

SparkR 支持通過 SparkDataFrame 接口對各種 data sources（數據源）進行操作. 本節介紹使用數據源加載和保存數據的常見方法. 您可以查看 Spark Sql 編程指南的 specific options 部分以了解更多可用於內置的 data sources（數據源）內容.

從數據源創建 SparkDataFrames 常見的方法是 read.df. 此方法將加載文件的路徑和數據源的類型，並且將自動使用當前活動的 SparkSession. SparkR 天生就支持讀取 JSON, CSV 和 Parquet 文件, 並且通過可靠來源的軟件包 第三方項目, 您可以找到 Avro 等流行文件格式的 data source connectors（數據源連接器）. 可以用 spark-submit 或 sparkR 命令指定 --packages 來添加這些包, 或者在交互式 R shell 或從 RStudio 中使用sparkPackages 參數初始化 SparkSession.

sparkR.session(sparkPackages = "com.databricks:spark-avro_2.11:3.0.0")

We can see how to use data sources using an example JSON input file. Note that the file that is used here is not a typical JSON file. Each line in the file must contain a separate, self-contained valid JSON object. For more information, please see JSON Lines text format, also called newline-delimited JSON. As a consequence, a regular multi-line JSON file will most often fail.

我們可以看看如何使用 JSON input file 的例子來使用數據源. 註意, 這裏使用的文件是 not 一個經典的 JSON 文件. 文件中的每行都必須包含一個單獨的，獨立的有效的JSON對象

people <- read.df("./examples/src/main/resources/people.json", "json")
head(people)
##  age    name
##1  NA Michael
##2  30    Andy
##3  19  Justin

# SparkR 自動從 JSON 文件推斷出 schema（模式）
printSchema(people)
# root
#  |-- age: long (nullable = true)
#  |-- name: string (nullable = true)

# 同樣, 使用  read.json 讀取多個文件
people <- read.json(c("./examples/src/main/resources/people.json", "./examples/src/main/resources/people2.json"))

該 data sources API 原生支持 CSV 格式的 input files（輸入文件）. 要了解更多信息請參閱 SparkR read.df API 文檔.

df <- read.df(csvPath, "csv", header = "true", inferSchema = "true", na.strings = "NA")

該 data sources API 也可用於將 SparkDataFrames 存儲為多個 file formats（文件格式）. 例如, 我們可以使用 write.df 把先前的示例的 SparkDataFrame 存儲為一個 Parquet 文件.

write.df(people, path = "people.parquet", source = "parquet", mode = "overwrite")

從 Hive tables 來創建 SparkDataFrame

您也可以從 Hive tables（表）來創建 SparkDataFrames. 為此，我們需要創建一個具有 Hive 支持的 SparkSession，它可以訪問 Hive MetaStore 中的 tables（表）. 請註意, Spark 應該使用 Hive support 來構建，更多細節可以在 SQL 編程指南 中查閱.

sparkR.session()

sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
sql("LOAD DATA LOCAL INPATH ‘examples/src/main/resources/kv1.txt‘ INTO TABLE src")

# Queries can be expressed in HiveQL.
results <- sql("FROM src SELECT key, value")

# results is now a SparkDataFrame
head(results)
##  key   value
## 1 238 val_238
## 2  86  val_86
## 3 311 val_311

SparkDataFrame 操作

SparkDataFrames 支持一些用於結構化數據處理的 functions（函數）. 這裏我們包括一些基本的例子，一個完整的列表可以在 API 文檔中找到:

Selecting rows（行）, columns（列）

# Create the SparkDataFrame
df <- as.DataFrame(faithful)

# 獲取關於 SparkDataFrame 基礎信息
df
## SparkDataFrame[eruptions:double, waiting:double]

# Select only the "eruptions" column
head(select(df, df$eruptions))
##  eruptions
##1     3.600
##2     1.800
##3     3.333

# You can also pass in column name as strings
head(select(df, "eruptions"))

# Filter the SparkDataFrame to only retain rows with wait times shorter than 50 mins
head(filter(df, df$waiting < 50))
##  eruptions waiting
##1     1.750      47
##2     1.750      47
##3     1.867      48

Grouping, Aggregation（分組, 聚合）

SparkR data frames 支持一些常見的, 用於在 grouping（分組）數據後進行 aggregate（聚合）的函數. 例如, 我們可以在 faithful dataset 中計算 waiting 時間的直方圖, 如下所示.

# We use the `n` operator to count the number of times each waiting time appears
head(summarize(groupBy(df, df$waiting), count = n(df$waiting)))
##  waiting count
##1      70     4
##2      67     1
##3      69     2

# We can also sort the output from the aggregation to get the most common waiting times
waiting_counts <- summarize(groupBy(df, df$waiting), count = n(df$waiting))
head(arrange(waiting_counts, desc(waiting_counts$count)))
##   waiting count
##1      78    15
##2      83    14
##3      81    13

Operating on Columns（列上的操作）

SparkR 還提供了一些可以直接應用於列進行數據處理和 aggregatation（聚合）的函數. 下面的例子展示了使用基本的算術函數.

# Convert waiting time from hours to seconds.
# Note that we can assign this to a new column in the same SparkDataFrame
df$waiting_secs <- df$waiting * 60
head(df)
##  eruptions waiting waiting_secs
##1     3.600      79         4740
##2     1.800      54         3240
##3     3.333      74         4440

應用 User-Defined Function（UDF 用戶自定義函數）

在 SparkR 中, 我們支持幾種 User-Defined Functions:

Run a given function on a large dataset using dapply or dapplyCollect

dapply

應用一個 function（函數）到 SparkDataFrame 的每個 partition（分區）. 應用於 SparkDataFrame 每個 partition（分區）的 function（函數）應該只有一個參數, 它中的 data.frame 對應傳遞的每個分區. 函數的輸出應該是一個 data.frame. Schema 指定生成的 SparkDataFrame row format. 它必須匹配返回值的 data types.

# Convert waiting time from hours to seconds.
# Note that we can apply UDF to DataFrame.
schema <- structType(structField("eruptions", "double"), structField("waiting", "double"),
                     structField("waiting_secs", "double"))
df1 <- dapply(df, function(x) { x <- cbind(x, x$waiting * 60) }, schema)
head(collect(df1))
##  eruptions waiting waiting_secs
##1     3.600      79         4740
##2     1.800      54         3240
##3     3.333      74         4440
##4     2.283      62         3720
##5     4.533      85         5100
##6     2.883      55         3300

dapplyCollect

像 dapply 那樣, 應用一個函數到 SparkDataFrame 的每個分區並且手機返回結果. 函數的輸出應該是一個 data.frame. 但是, 不需要傳遞 Schema. 註意, 如果運行在所有分區上的函數的輸出不能 pulled（拉）到 driver 的內存中過去, 則 dapplyCollect 會失敗.

# Convert waiting time from hours to seconds.
# Note that we can apply UDF to DataFrame and return a R‘s data.frame
ldf <- dapplyCollect(
         df,
         function(x) {
           x <- cbind(x, "waiting_secs" = x$waiting * 60)
         })
head(ldf, 3)
##  eruptions waiting waiting_secs
##1     3.600      79         4740
##2     1.800      54         3240
##3     3.333      74         4440

Run a given function on a large dataset grouping by input column(s) and using gapply or gapplyCollect（在一個大的 dataset 上通過 input colums（輸入列）來進行 grouping（分組）並且使用 gapply or gapplyCollect 來運行一個指定的函數）

gapply

應用給一個函數到 SparkDataFrame 的每個 group. 該函數被應用到 SparkDataFrame 的每個 group, 並且應該只有兩個參數: grouping key 和 R data.frame 對應的 key. 該 groups 從 SparkDataFrame 的 columns（列）中選擇. 函數的輸出應該是 data.frame. Schema 指定生成的 SparkDataFrame row format. 它必須在 Spark data types 數據類型 的基礎上表示 R 函數的輸出 schema（模式）. 用戶可以設置返回的 data.frame列名.

# Determine six waiting times with the largest eruption time in minutes.
schema <- structType(structField("waiting", "double"), structField("max_eruption", "double"))
result <- gapply(
    df,
    "waiting",
    function(key, x) {
        y <- data.frame(key, max(x$eruptions))
    },
    schema)
head(collect(arrange(result, "max_eruption", decreasing = TRUE)))

##    waiting   max_eruption
##1      64       5.100
##2      69       5.067
##3      71       5.033
##4      87       5.000
##5      63       4.933
##6      89       4.900

gapplyCollect

像 gapply 那樣, 將函數應用於 SparkDataFrame 的每個分區，並將結果收集回 R data.frame. 函數的輸出應該是一個 data.frame. 但是，不需要傳遞 schema（模式）. 請註意，如果在所有分區上運行的 UDF 的輸出無法 pull（拉）到 driver 的內存, 那麽 gapplyCollect 可能會失敗.

# Determine six waiting times with the largest eruption time in minutes.
result <- gapplyCollect(
    df,
    "waiting",
    function(key, x) {
        y <- data.frame(key, max(x$eruptions))
        colnames(y) <- c("waiting", "max_eruption")
        y
    })
head(result[order(result$max_eruption, decreasing = TRUE), ])

##    waiting   max_eruption
##1      64       5.100
##2      69       5.067
##3      71       5.033
##4      87       5.000
##5      63       4.933
##6      89       4.900

使用 spark.lapply 分發運行一個本地的 R 函數

spark.lapply

類似於本地 R 中的 lapply, spark.lapply 在元素列表中運行一個函數，並使用 Spark 分發計算. 以類似於 doParallel 或 lapply 的方式應用於列表的元素. 所有計算的結果應該放在一臺機器上. 如果不是這樣, 他們可以像 df < - createDataFrame(list) 這樣做, 然後使用 dapply.

# Perform distributed training of multiple models with spark.lapply. Here, we pass
# a read-only list of arguments which specifies family the generalized linear model should be.
families <- c("gaussian", "poisson")
train <- function(family) {
  model <- glm(Sepal.Length ~ Sepal.Width + Species, iris, family = family)
  summary(model)
}
# Return a list of model‘s summaries
model.summaries <- spark.lapply(families, train)

# Print the summary of each model
print(model.summaries)

SparkR 中運行 SQL 查詢

A SparkDataFrame can also be registered as a temporary view in Spark SQL and that allows you to run SQL queries over its data. The sqlfunction enables applications to run SQL queries programmatically and returns the result as a SparkDataFrame.

# Load a JSON file
people <- read.df("./examples/src/main/resources/people.json", "json")

# Register this SparkDataFrame as a temporary view.
createOrReplaceTempView(people, "people")

# SQL statements can be run by using the sql method
teenagers <- sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")
head(teenagers)
##    name
##1 Justin

機器學習

算法

SparkR 現支持下列機器學習算法:

分類

• spark.logit: 邏輯回歸 Logistic Regression
• spark.mlp: 多層感知 (MLP)
• spark.naiveBayes: 樸素貝葉斯
• spark.svmLinear: 線性支持向量機

回歸

• spark.survreg: 加速失敗時間生存模型 Accelerated Failure Time (AFT) Survival Model
• spark.glm or glm: 廣義線性模型 Generalized Linear Model (GLM)
• spark.isoreg: 保序回歸

樹

• spark.gbt: 梯度提升樹 for 回歸 and 分類
• spark.randomForest: 隨機森林 for 回歸 and 分類

聚類

• spark.bisectingKmeans: 二分k均值
• spark.gaussianMixture: 高斯混合模型 (GMM)
• spark.kmeans: K-Means
• spark.lda: 隱含狄利克雷分布 (LDA)

協同過濾

• spark.als: 交替最小二乘 (ALS)

頻繁模式挖掘

• spark.fpGrowth : FP-growth

統計

• spark.kstest: 柯爾莫哥洛夫-斯米爾諾夫檢驗

SparkR 底層實現使用 MLlib 來訓練模型. 有關示例代碼，請參閱MLlib用戶指南的相應章節. 用戶可以調用summary輸出擬合模型的摘要, 利用模型對數據進行預測, 並且使用 write.ml/read.ml 來 保存/加載擬合的模型 . SparkR 支持對模型擬合使用部分R的公式運算符, 包括 ‘~’, ‘.’, ‘:’, ‘+’, 和 ‘-‘.

模型持久化

下面的例子展示了SparkR如何 保存/加載 機器學習模型.

training <- read.df("data/mllib/sample_multiclass_classification_data.txt", source = "libsvm")
# Fit a generalized linear model of family "gaussian" with spark.glm
df_list <- randomSplit(training, c(7,3), 2)
gaussianDF <- df_list[[1]]
gaussianTestDF <- df_list[[2]]
gaussianGLM <- spark.glm(gaussianDF, label ~ features, family = "gaussian")

# Save and then load a fitted MLlib model
modelPath <- tempfile(pattern = "ml", fileext = ".tmp")
write.ml(gaussianGLM, modelPath)
gaussianGLM2 <- read.ml(modelPath)

# Check model summary
summary(gaussianGLM2)

# Check model prediction
gaussianPredictions <- predict(gaussianGLM2, gaussianTestDF)
head(gaussianPredictions)

unlink(modelPath)

R和Spark之間的數據類型映射

Structured Streaming

SparkR 支持 Structured Streaming API (測試階段). Structured Streaming 是一個 構建於SparkSQL引擎之上的易拓展、可容錯的流式處理引擎. 更多信息請參考 R API Structured Streaming Programming Guide

R 函數名沖突

當在R中加載或引入(attach)一個新package時, 可能會發生函數名沖突,一個函數掩蓋了另一個函數

下列函數是被SparkR所掩蓋的:

stats::cov(x, y = NULL, use = "everything",
           method = c("pearson", "kendall", "spearman"))

stats::filter(x, filter, method = c("convolution", "recursive"),
              sides = 2, circular = FALSE, init)

由於SparkR的一部分是在dplyr軟件包上建模的，因此SparkR中的某些函數與dplyr中同名. 根據兩個包的加載順序, 後加載的包會掩蓋先加載的包的部分函數. 在這種情況下, 可以在函數名前指定包名前綴, 例如: SparkR::cume_dist(x) or dplyr::cume_dist(x).

你可以在 R 中使用search()檢查搜索路徑

遷移指南

SparkR 1.5.x 升級至 1.6.x

• 在Spark 1.6.0 之前, 寫入模式默認值為 append. 在 Spark 1.6.0 改為 error 匹配 Scala API.
• SparkSQL 將R 中的 NA 轉換為 null,反之亦然.

SparkR 1.6.x 升級至 2.0

• table 方法已經移除並替換為 tableToDF.
• 類 DataFrame 已改名為 SparkDataFrame 避免名稱沖突.
• Spark的 SQLContext 和 HiveContext 已經過時並替換為 SparkSession. 相應的摒棄 sparkR.init()而通過調用 sparkR.session() 來實例化SparkSession. 一旦實例化完成, 當前的SparkSession即可用於SparkDataFrame 操作(註釋:spark2.0開始所有的driver實例通過sparkSession來進行構建).
• sparkR.session 不支持 sparkExecutorEnv 參數.要為executors設置環境，請使用前綴”spark.executorEnv.VAR_NAME”設置Spark配置屬性，例如”spark.executorEnv.PATH”, -sqlContext 不再需要下列函數: createDataFrame, as.DataFrame, read.json, jsonFile, read.parquet, parquetFile, read.text, sql, tables, tableNames, cacheTable, uncacheTable, clearCache, dropTempTable, read.df, loadDF, createExternalTable.
• registerTempTable 方法已經過期並且替換為createOrReplaceTempView.
• dropTempTable 方法已經過期並且替換為 dropTempView.
• sc SparkContext 參數不再需要下列函數: setJobGroup, clearJobGroup, cancelJobGroup

升級至 SparkR 2.1.0

• join 不再執行笛卡爾積計算, 使用 crossJoin 來進行笛卡爾積計算.

升級至 SparkR 2.2.0

• createDataFrame 和 as.DataFrame 添加numPartitions參數. 數據分割時, 分區位置計算已經與scala計算相一致.
• 方法 createExternalTable 已經過期並且替換為createTable. 可以調用這兩種方法來創建外部或托管表. 已經添加額外的 catalog 方法.
• 默認情況下，derby.log現在已保存到tempdir()目錄中. 當實例化SparkSession且選項enableHiveSupport 為TRUE,會創建derby.log .
• 更正spark.lda 錯誤設置優化器的bug.
• 更新模型概況輸出 coefficients as matrix. 更新的模型概況包括 spark.logit, spark.kmeans, spark.glm. spark.gaussianMixture 的模型概況已經添加對數概度(log-likelihood) loglik.

我們一直在努力

apachecn/spark-doc-zh

原文地址: http://spark.apachecn.org/docs/cn/2.2.0/sparkr.html
網頁地址: http://spark.apachecn.org/
github: https://github.com/apachecn/spark-doc-zh（覺得不錯麻煩給個 Star，謝謝！~）

Apache Spark 2.2.0 中文文檔 - SparkR (R on Spark) | ApacheCN