快速開始使用spark

一、版本說明

• 在spark2.0版本之前，spakr編程接口是RDD(Resilient Distributed Dataset,彈性分佈式數據集)，spark2.0版本即以上，RDD被Dataset取代，Dataset比RDD更爲強大，在底層獲得了許多優化了。固然2.0+版本仍然支持RDD，但官方建議使用Dataset。

二、安全

• spark的安全模式默認是關閉的，這意味着你可能收到攻擊。

三、利用Spark Shell進行交互式數據分析

• Spark的shell提供了一種學習API的簡單方法，以及一種以交互方式分析數據的強大工具。
• 能夠經過使用scala或者python進行編程。
• 在spark的安裝根目錄下啓動。

3.一、Scala方式

啓動 
./bin/spark-shell
讀取一個文件用來建立一個新的數據集Dataset
val textFile = spark.read.textFile("README.md")

對數據集進行操做
textFile.count()
textFile.first()
val linesWithSpark = textFile.filter(line => line.contains("Spark"))
textFile.filter(line => line.contains("Spark")).count()

3.二、python方式

啓動
./bin/pyspark
textFile = spark.read.text("README.md")
textFile.count()
textFile.first()
linesWithSpark = textFile.filter(textFile.value.contains("Spark"))
textFile.filter(textFile.value.contains("Spark")).count()

四、Dataset的更多操做

1.查找文件中長度最大的字符串，並返回長度
textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)

2.實現wordcounts
val wordCounts = textFile.flatMap(line => line.split(" ")).groupByKey(identity).count()
wordCounts.collect()

五、緩存Caching

• Spark還支持將數據集提取到羣集範圍的內存緩存中。這在重複訪問數據時很是有用，例如查詢小的「熱」數據集或運行像PageRank這樣的迭代算法時。舉個簡單的例子，讓咱們標記linesWithSpark要緩存的數據集：

linesWithSpark.cache()
linesWithSpark.count()

經過文件運行

• 新建一個SimpleApp.scala

/* SimpleApp.scala */
import org.apache.spark.sql.SparkSession

object SimpleApp {
  def main(args: Array[String]) {
    val logFile = "YOUR_SPARK_HOME/README.md" // Should be some file on your system
    val spark = SparkSession.builder.appName("Simple Application").getOrCreate()
    val logData = spark.read.textFile(logFile).cache()
    val numAs = logData.filter(line => line.contains("a")).count()
    val numBs = logData.filter(line => line.contains("b")).count()
    println(s"Lines with a: $numAs, Lines with b: $numBs")
    spark.stop()
  }
}

• 運行結果