1、Apache Spark-3福建師大數(shù)信學院 張仕目 錄C o n t e n t s01020304Spark簡介Spark安裝與應用Spark API實例Spark的運行機制Spark API實例Ø Spark的APIØ DataFrame API實例Ø RDD API實例Ø MLlib實例1 Spark的APIØ Spark是基于分布式數(shù)據(jù)集的概念的，可以包含任意的Java、Python對象。我們只需要基于這些外部數(shù)據(jù)構造數(shù)據(jù)集，然后對這些數(shù)據(jù)集進行并行操作。Spark API的基礎構件是RDD API，在RDD API之上，又提供了 的

2、API 供使用，例如DataFrame API， 學習API。這些更 次的API提供了特定數(shù)據(jù)操作的方法，本部分將通過若干例子說明最簡單的Spark應用，展示Spark的強大功能。2 RDD API實例 Word Count：構造 (String, Int) 數(shù)據(jù)集，并保存。Java：JavaRDD<String> textFile = sc.textFile("hdfs:/."); JavaPairRDD<String, Integer> counts = textFile.flatMap(s -> Arrays.asList(s.split

3、(" ").iterator().mapToPair(word -> new Tuple2<>(word, 1).reduceByKey(a, b) -> a + b); counts.saveAsTextFile("hdfs:/.");Scala：val textFile = sc.textFile("hdfs:/.")val counts = textFile.flatMap(line => line.split(" ").map(word => (word, 1).redu

4、ceByKey(_ + _) counts.saveAsTextFile("hdfs:/.")Python：text_file = sc.textFile("hdfs:/.")counts = text_file.flatMap(lambda line: line.split(" ") .map(lambda word: (word, 1) .reduceByKey(lambda a, b: a + b) counts.saveAsTextFile("hdfs:/.")3 DataFrame API實例textFi

5、le = sc.textFile("hdfs:/.")# Creates a DataFrame having a single column named "line"df = textFile.map(lambda r: Row(r).toDF("line")errors = df.filter(col("line").like("%ERROR%") # Counts all the errorserrors.count()# Counts errors mentioning MySQL er

6、rors.filter(col("line").like("%MySQL%").count() # Fetches the MySQL errors as an array of strings errors.filter(col("line").like("%MySQL%").collect()Ø DataFrame API實例： DataFrame是分布式數(shù)據(jù)集，其數(shù)據(jù)帶列名組織。DataFrame API可以方便的支持各種關系操作。并且，基于DataFrame API的程序會自動使用Spark內置優(yōu)

7、化器進行優(yōu)化。Ø 實例：文本查找（Python）3 DataFrame API實例 Ø 實例：文本查找（Scala） val textFile = sc.textFile("hdfs:/.")/ Creates a DataFrame having a single column named "line"val df = textFile.toDF("line")val errors = df.filter(col("line").like("%ERROR%")/ Count

8、s all the errors errors.count()/ Counts errors mentioning MySQL errors.filter(col("line").like("%MySQL%").count()/ Fetches the MySQL errors as an array of strings errors.filter(col("line").like("%MySQL%").collect()3 DataFrame API實例 Ø 實例：文本查找（Java） / Creat

9、es a DataFrame having a single column named "line"JavaRDD<String> textFile = sc.textFile("hdfs:/."); JavaRDD<Row> rowRDD = textFile.map(RowFactory:create); List<StructField> fields = Arrays.asList( DataTypes.createStructField("line", DataTypes.StringTy

10、pe, true); StructType schema = DataTypes.createStructType(fields); DataFrame df = sqlContext.createDataFrame(rowRDD, schema);DataFrame errors = df.filter(col("line").like("%ERROR%");errors.count();/ Counts all the errors/ Counts errors mentioning MySQL errors.filter(col("lin

11、e").like("%MySQL%").count();/ Fetches the MySQL errors as an array of strings errors.filter(col("line").like("%MySQL%").collect();4 MLlib實例# Every record of this DataFrame contains the label and# features represented by a vector.df = sqlContext.createDataFrame(data

12、, "label", "features")# Set parameters for the algorithm.# Here, we limit the number of iterations to 10. lr = LogisticRegression(maxIter=10)m= lr.fit(df) # Fit the mto the data.# Given a dataset, predict each point's label, and show the results.m.transform(df).show()實例：回歸（Pr

13、ediction with Logistic Regression）（Python）本實例中，利用一些帶 的數(shù)據(jù)集和特征向量，利用回歸算法進行數(shù)據(jù) 的預測。Machine Learning 實例：MLlib（Sparks Machine Learning (ML) library）提供了許多分布式ML算法。這些算法覆蓋了特征提取、分類、回歸、聚類、推薦等等方面。MLlib 也提供一些諸如ML Pipeline構造工作流、CrossValidator用于調參、模型持久性用于和裝載模型。4 MLlib實例/ Every record of this DataFrame contains the l

14、abel and/ features represented by a vector.val df = sqlContext.createDataFrame(data).toDF("label", "features")/ Set parameters for the algorithm./ Here, we limit the number of iterations to 10. val lr = new LogisticRegression().setMaxIter(10) 些帶的數(shù)據(jù)集 和特征向量，利用回歸算法進行數(shù)據(jù)/ Fit the mto

15、the data.val m= lr.fit(df)的。/ Inspect the mval weights = m: get the feature weights.weights/ Given a dataset, predict each point's label, and show the results.m.transform(df).show()實例：回歸（Prediction with Logistic Regression）（Scala）本實例中，利用一4 MLlib實例/ Every record of this DataFrame contains the lab

16、el and/ features represented by a vector.StructType schema = new StructType(new StructFieldnew StructField("label", DataTypes.DoubleType, false, Metadata.empty(), new StructField("features", new VectorUDT(), false, Metadata.empty(),);DataFrame df = jsql.createDataFrame(data, sche

17、ma);/ Set parameters for the algorithm./ Here, we limit the number of iterations to 10. LogisticRegression lr = new LogisticRegression().setMaxIter(10); 些帶的數(shù)據(jù)集 和特征向量，利用回歸算法進行數(shù)據(jù)LogisticRegressionMm= lr.fit(df); / Fit the mto the data.的。/ Inspect the mVector weights = m: get the feature weights.weight

18、s();/ Given a dataset, predict each point's label, and show the results.m.transform(df).show();實例：回歸（Prediction with Logistic Regression）（Java）本實例中，利用一Spark安裝與應用Ø Spark的安裝scala程序開發(fā)環(huán)境與步驟1 Spark的安裝推薦：把Spark安裝于Linux系統(tǒng)。具體安裝 Apache Spark 的步驟如下：Step 1: 驗證Java是否正確安裝$java -version若正確輸出Java版本信息，則說明J

19、ava已經正確安裝，否則需要安裝Java， 并設置相關路徑。Step 2: 驗證Scala是否正確安裝$scala -version若正確輸出Scala版本信息，則說明Scala正確安裝，可跳過Step3和Step4，否則需要安裝Scala。Step 3:Scala最新版Scala，這里使用的是 scala-2.11.6 。1 Spark的安裝Step 4: 安裝 Scala解壓tar文件，命令如下：$ tar xvf scala-2.11.6.tgz把解壓后的文件移到/usr/local/scala（也可以是你指定的其他文件夾位置）$ su Password:# cd /home/Hadoo

20、p/Downloads/ # mv scala-2.11.6 /usr/local/scala # exit設置Scala路徑：$ export PATH = $PATH:/usr/local/scala/bin驗證安裝是否$scala -version：正確顯示Scala版本信息，則說明安裝。1 Spark的安裝Step 5:Apache Spark到Apache Spark官網Spark，本實例這里的是spark-1.3.1-bin-hadoop2.6Step 6: 安裝Spark按解壓、修改存放文件夾、設置環(huán)境變量幾個步驟進行Spark的設置，參考如下：$ tar xvf spark-1.3.1-bin-hadoop2.6.tgz$ su Password:# cd /home/Hadoop/Downloads/# mv spark-1.3.1-bin-hadoop2.6 /usr/local/spark # exit設置環(huán)境變量：打開 /.bashrc文件，并且在文件后面加上如下路徑： export PATH = $PATH:/usr/local/spark/bin執(zhí)行 /.bashrc file，使設置生效。$ sou