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 中进行设置:

df <- as.DataFrame(faithful)

# 展示第一个 SparkDataFrame 的内容

head(df)

##  eruptions waiting

##1     3.600      79

##2     1.800      54

##3     3.333      74

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

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"))

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

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

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

# 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

# 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

# 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

# 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

# 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

# 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

# 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

# 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)

# 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

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)