00、Word Count

時間 2019-11-05

標籤 word count 欄目 Microsoft Office 简体版

原文原文鏈接

一、開發環境

一、eclipse-jee-neon-3

二、scala-ide： http://download.scala-ide.org/sdk/lithium/e46/scala212/stable/site

三、下載Maven ，修改D:\apache-maven-3.5.0\conf\settings.xml配置文件：

本地庫位置：

<localRepository>D:/apache-maven-3.5.0/localRepository</localRepository>

倉庫使用阿里雲：

<id>nexus-aliyun</id>

<name>Nexus aliyun</name>

<url>http://maven.aliyun.com/nexus/content/groups/public</url>

</mirror>

四、配置Maven

二、本地Local模擬運行

本地模式，即不須要啓動Spark便可模擬運行，若是初始RDD來自於HDFS上的文件，則僅需啓動Hadoop，不論是在Eclipse中直接運行，仍是經過 spark-submit提交運行

2.一、Java版

pom.xml文件：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

<groupId>sparkcore</groupId>

<artifactId>sparkcore-java</artifactId>

<name>sparkcore-java</name>

<url>http://maven.apache.org</url>

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

</properties>

<groupId>org.apache.spark</groupId>

<artifactId>spark-core_2.11</artifactId>

<scope>compile</scope>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-sql_2.11</artifactId>

<scope>compile</scope>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-hive_2.11</artifactId>

<scope>compile</scope>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-streaming_2.11</artifactId>

<scope>compile</scope>

</dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-client</artifactId>

<scope>compile</scope>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-streaming-kafka_2.11</artifactId>

<scope>compile</scope>

</dependency>

<groupId>com.fasterxml.jackson.core</groupId>

<artifactId>jackson-core</artifactId>

</dependency>

</dependencies>

</project>

package sparkcore.java;

import java.util.Arrays;

import java.util.Iterator;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaPairRDD;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.function.FlatMapFunction;

import org.apache.spark.api.java.function.Function2;

import org.apache.spark.api.java.function.PairFunction;

import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

/**

* 使用java開發本地測試的wordcount程序

public class WordCountLocal {

public static void main(String[] args) {

// 編寫Spark應用程序

// 本地執行，是能夠執行在eclipse中的main方法中，執行的

// 第一步：建立SparkConf對象，設置Spark應用的配置信息

// 使用setMaster()能夠設置Spark應用程序要鏈接的Spark集羣的master節點的url

// 可是若是設置爲local則表明，在本地運行

SparkConf conf = new SparkConf().setAppName("WordCountLocal").setMaster("local");

// 以spark-submit提交運行時，須要設置,如直接以Java應用程序啓動，則能夠去掉

conf.set("spark.testing.memory", "536870912");// 512M

// 第二步：建立JavaSparkContext對象

// 在Spark中，SparkContext是Spark全部功能的一個入口，你不管是用java、scala，甚至是python編寫

// 都必需要有一個SparkContext，它的主要做用，包括初始化Spark應用程序所需的一些核心組件，包括

// 調度器（DAGSchedule、TaskScheduler），還會去到Spark Master節點上進行註冊，等等

// 一句話，SparkContext，是Spark應用中，能夠說是最最重要的一個對象

// 可是呢，在Spark中，編寫不一樣類型的Spark應用程序，使用的SparkContext是不一樣的，若是使用scala，

// 使用的就是原生的SparkContext對象

// 可是若是使用Java，那麼就是JavaSparkContext對象

// 若是是開發Spark SQL程序，那麼就是SQLContext、HiveContext

// 若是是開發Spark Streaming程序，那麼就是它獨有的SparkContext

// 以此類推

JavaSparkContext sc = new JavaSparkContext(conf);

// 第三步：要針對輸入源（hdfs文件、本地文件，等等），建立一個初始的RDD

// 輸入源中的數據會打散，分配到RDD的每一個partition中，從而造成一個初始的分佈式的數據集

// 咱們這裏呢，由於是本地測試，因此呢，就是針對本地文件

// SparkContext中，用於根據文件類型的輸入源建立RDD的方法，叫作textFile()方法

// 在Java中，建立的普通RDD，都叫作JavaRDD

// 在這裏呢，RDD中，有元素這種概念，若是是hdfs或者本地文件呢，建立的RDD，每個元素就至關於是文件裏的一行

// JavaRDD<String> lines = sc.textFile("file:///D:/eclipse-jee-neon-3/workspace/sparkcore-java/test.txt");

// 若是本地文件爲Linux

JavaRDD<String> lines = sc.textFile("file:////root/spark/core/test.txt");

// 若是文件放在HDFS上時，須要先啓動Hadoop

// JavaRDD<String> lines = sc.textFile("hdfs://node1:8020/test.txt");

// 第四步：對初始RDD進行transformation操做，也就是一些計算操做

// 一般操做會經過建立function，並配合RDD的map、flatMap等算子來執行

// function，一般，若是比較簡單，則建立指定Function的匿名內部類

// 可是若是function比較複雜，則會單首創建一個類，做爲實現這個function接口的類

// 先將每一行拆分紅單個的單詞

// FlatMapFunction，有兩個泛型參數，分別表明了輸入和輸出類型

// 咱們這裏呢，輸入確定是String，由於是一行一行的文本，輸出，其實也是String，只不過是多個放在集合中

// 這裏先簡要介紹flatMap算子的做用，其實就是，將RDD的一個元素，給拆分紅一個或多個元素

JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

private static final long serialVersionUID = 1L;

public Iterator<String> call(String line) throws Exception {

return Arrays.asList(line.split(" ")).iterator();

}

});

// 接着，須要將每個單詞，映射爲(單詞, 1)的這種格式

// 由於只有這樣，後面才能根據單詞做爲key，來進行每一個單詞的出現次數的累加

// mapToPair，其實就是將每一個元素，映射爲一個(v1,v2)這樣的Tuple2類型的元素

// 若是你們還記得scala裏面講的tuple，那麼沒錯，這裏的tuple2就是scala類型，包含了兩個值

// mapToPair這個算子，要求的是與PairFunction配合使用，第一個泛型參數表明了輸入類型

// 第二個和第三個泛型參數，表明的輸出的Tuple2的第一個值和第二個值的類型

// JavaPairRDD的兩個泛型參數，分別表明了tuple元素的第一個值和第二個值的類型

JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {

private static final long serialVersionUID = 1L;

public Tuple2<String, Integer> call(String word) throws Exception {

return new Tuple2<String, Integer>(word, 1);

}

});

// 接着，須要以單詞做爲key，統計每一個單詞出現的次數

// 這裏要使用reduceByKey這個算子，對每一個key對應的value，都進行reduce操做

// 好比JavaPairRDD中有幾個元素，分別爲(hello, 1) (hello, 1) (hello, 1) (world, 1)

// reduce操做，至關因而把第一個值和第二個值進行計算，而後再將結果與第三個值進行計算

// 好比這裏的hello，那麼就至關因而，首先是1 + 1 = 2，而後再將2 + 1 = 3

// 最後返回的JavaPairRDD中的元素，也是tuple，可是第一個值就是每一個key，第二個值就是key的value

// reduce以後的結果，至關於就是每一個單詞出現的次數

JavaPairRDD<String, Integer> wordCounts = pairs.reduceByKey(

// 第一與第二個參數爲輸入類型（爲兩個Tuple2的第二個元素類型），第三個爲輸出類型

new Function2<Integer, Integer, Integer>() {

private static final long serialVersionUID = 1L;

public Integer call(Integer v1, Integer v2) throws Exception {

return v1 + v2;

}

});

// 到這裏爲止，咱們經過幾個Spark算子操做，已經統計出了單詞的次數

// 可是，以前咱們使用的flatMap、mapToPair、reduceByKey這種操做，都叫作transformation操做

// 一個Spark應用中，光是有transformation操做，是不行的，是不會執行的，必需要有一種叫作action

// 接着，最後，可使用一種叫作action操做的，好比說，foreach，來觸發程序的執行

wordCounts.foreach(new VoidFunction<Tuple2<String, Integer>>() {

private static final long serialVersionUID = 1L;

public void call(Tuple2<String, Integer> wordCount) throws Exception {

System.out.println(wordCount._1 + " : " + wordCount._2);

}

});

sc.close();

}

將上面程序經過Mave打包成jar文件，上傳到Linux平臺，而後經過 spark-submit提交本地運行（無需啓動Spark，若是用到的文件在HDFS上面，則要啓動Hadoop）：

spark-submit \

--master local[1] \

--class sparkcore.java.WordCountLocal \

--num-executors 1 \

--driver-memory 100m \

--executor-memory 100m \

--executor-cores 2 \

/root/spark/core/sparkcore-java-1.0.jar

2.二、配置Hadoop客戶端

運行時可能會拋以下異常，但不影響結果，緣由是Hadoop沒有配置，也能夠按照下面進行配置

一、將hadoop-2.7.3.tar.gz（前面本身編譯的CentOS版本）解壓到D:\hadoop-2.7.3，並將winutils.exe、hadoop.dll等文件到 D:\hadoop-2.7.3\bin 下，再將hadoop.dll放到C:\Windows及C:\Windows\System32下

二、添加HADOOP_HOME環境變量，值爲D:\hadoop\hadoop-2.7.2，並將%HADOOP_HOME%\bin添加到Path環境變量中

三、雙擊winutils.exe，若是出現「缺失MSVCR120.dll」的提示，則安裝VC++2013相關組件

四、將hadoop-eclipse-plugin-2.7.3.jar插件包拷貝到Eclipse plugins 目錄下

五、運行Eclipse，進行配置：

2.三、Scala版：

一、先建立Scala普通工程

二、右擊工程，將普通工程轉換爲Maven工程

三、 pom.xm

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

<groupId>sparkcore</groupId>

<artifactId>sparkcore-java</artifactId>

<name>sparkcore-scala</name>

<url>http://maven.apache.org</url>

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

</properties>

<groupId>org.apache.spark</groupId>

<artifactId>spark-core_2.11</artifactId>

<scope>compile</scope>

<groupId>org.scala-lang</groupId>

<artifactId>scala-library</artifactId>

</exclusion>

</exclusions>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-sql_2.11</artifactId>

<scope>compile</scope>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-hive_2.11</artifactId>

<scope>compile</scope>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-streaming_2.11</artifactId>

<scope>compile</scope>

<groupId>org.scala-lang</groupId>

<artifactId>scala-library</artifactId>

</exclusion>

</exclusions>

</dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-client</artifactId>

<scope>compile</scope>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-streaming-kafka_2.11</artifactId>

<scope>compile</scope>

<groupId>org.scala-lang</groupId>

<artifactId>scala-library</artifactId>

</exclusion>

</exclusions>

</dependency>

<groupId>com.fasterxml.jackson.core</groupId>

<artifactId>jackson-core</artifactId>

</dependency>

</dependencies>

<build>

</build>

</project>

四、將scala版本修改爲2.11：

五、而後打包與Java版本的Maven同樣，也能夠打包jar包

package sparkcore.scala

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

object WordCountLocal {

def main(args: Array[String]) {

val conf = new SparkConf().setAppName("WordCount").setMaster("local")

conf.set("spark.testing.memory", "536870912"); // 512M

val sc = new SparkContext(conf)

// val lines = sc.textFile("file:///D:/eclipse-jee-neon-3/workspace/sparkcore-scala/test.txt", 1);

// val lines = sc.textFile("file:////root/spark/core/test.txt", 1);

val lines = sc.textFile("hdfs://node1:8020/test.txt", 1);

val words = lines.flatMap { line => line.split(" ") }

val pairs = words.map { word => (word, 1) }

val wordCounts = pairs.reduceByKey { _ + _ }

wordCounts.foreach(wordCount => println(wordCount._1 + " : " + wordCount._2))

}

與Java版本同樣，能夠在Eclipse中直接運行，也能夠打成jar，傳到Linux上，經過 spark-submit提交本地運行：

spark-submit \

--master local[1] \

--class sparkcore.scala.WordCountLocal \

--num-executors 1 \

--driver-memory 100m \

--executor-memory 100m \

--executor-cores 2 \

/root/spark/core/sparkcore-scala-1.0.jar

三、Standalone模式運行

將任務提交到Spark集羣上運行，因此Spark必定要開啓，並有屏幕打印將不會直接打屏，而是在相應Work後臺日誌中顯示

3.一、Java版

spark-submit \

--master spark://node1:7077 \

--class sparkcore.java.WordCountCluster \

--num-executors 1 \

--driver-memory 100m \

--executor-memory 512m \

--executor-cores 1 \

/root/spark/core/sparkcore-java-1.0.jar

package sparkcore.java;

import java.util.Arrays;

import java.util.Iterator;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaPairRDD;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.function.FlatMapFunction;

import org.apache.spark.api.java.function.Function2;

import org.apache.spark.api.java.function.PairFunction;

import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

/**

* 將java開發的wordcount程序部署到spark集羣上運行

public class WordCountCluster {

public static void main(String[] args) {

// 若是要在spark集羣上運行，須要修改的，只有兩個地方

// 第一，將SparkConf的setMaster()方法給刪掉，默認它本身會去鏈接（未設置時運行時以spark-submit中的--master指定模式爲準）

// 第二，咱們針對的不是本地文件了，修改成hadoop hdfs上的真正的存儲大數據的文件

// 實際執行步驟：

// 一、將test.txt文件上傳到hdfs上去

// 二、使用maven插件，對spark工程進行打Jar包

// 三、將打包後的spark工程jar包，上傳到機器上執行

// 四、編寫spark-submit腳本

// 五、執行spark-submit腳本，提交spark應用到集羣執行

SparkConf conf = new SparkConf().setAppName("WordCountCluster");

conf.set("spark.testing.memory", "536870912");// 512M

JavaSparkContext sc = new JavaSparkContext(conf);

JavaRDD<String> lines = sc.textFile("hdfs://node1:8020/test.txt"); //注：也能夠是操做系統本地文件，請看後面Scala版本

JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

private static final long serialVersionUID = 1L;

public Iterator<String> call(String line) throws Exception {

return Arrays.asList(line.split(" ")).iterator();

}

});

JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {

private static final long serialVersionUID = 1L;

public Tuple2<String, Integer> call(String word) throws Exception {

return new Tuple2<String, Integer>(word, 1);

}

});

JavaPairRDD<String, Integer> wordCounts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {

private static final long serialVersionUID = 1L;

public Integer call(Integer v1, Integer v2) throws Exception {

return v1 + v2;

}

});

wordCounts.foreach(new VoidFunction<Tuple2<String, Integer>>() {

private static final long serialVersionUID = 1L;

public void call(Tuple2<String, Integer> wordCount) throws Exception {

System.out.println(wordCount._1 + " : " + wordCount._2);

}

});

sc.close();

}

3.二、Scala版

spark-submit \

--master spark://node1:7077 \

--class sparkcore.scala.WordCountCluster \

--num-executors 1 \

--driver-memory 100m \

--executor-memory 512m \

--executor-cores 1 \

/root/spark/core/sparkcore-scala-1.0.jar

package sparkcore.scala

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

object WordCountCluster {

def main(args: Array[String]) {

//當程序中使用setMaster指定後，以程序中設定的爲準，會忽略掉spark-submit的--master

//若是沒有設置setMaster，則以spark-submit中的--master爲準

val conf = new SparkConf().setAppName("WordCountCluster").setMaster("spark://node1:7077")

conf.set("spark.testing.memory", "536870912") //

val sc = new SparkContext(conf)

val lines = sc.textFile("hdfs://node1:8020/test.txt")

//若是以Standalone模式運行時，若是讀取的操做系統本地文件，則每一個Work上都要拷貝一份（但只會讀取某一臺機器上的文件）

// val lines = sc.textFile("file:////root/spark/core/test.txt", 1);

val words = lines.flatMap { line => line.split(" ") }

val pairs = words.map { word => (word, 1) }

val wordCounts = pairs.reduceByKey { _ + _ }

wordCounts.foreach(wordCount => println(wordCount._1 + " : " + wordCount._2))

}

Spark配置項的優先順序

（1）在用戶代碼中用SparkConf對象上的set（）函數顯式聲明的配置。

（2）傳遞給spark-submit或spark-shell的標誌。

（3）在spark-defaults.conf屬性文件中的值。

（4）Spark的默認值。

四、在spark-shell中運行

運行前，須要啓動Spark。 spark-shell至關於本地Local模式，不會提交到Spark集羣上跑，所以若是用的本地文件，則只須要在運行spark-shell 機器本地上放一個文件便可

[root@node1 ~]# spark-shell

scala> :paste

// Entering paste mode (ctrl-D to finish)

val lines = sc.textFile("file:////root/spark/core/test.txt", 1);

val words = lines.flatMap { line => line.split(" ") }

val pairs = words.map { word => (word, 1) }

val wordCounts = pairs.reduceByKey { _ + _ }

wordCounts.foreach(wordCount => println(wordCount._1 + " : " + wordCount._2))

// Exiting paste mode, now interpreting.

you : 2

hello : 4

me : 2

lines: org.apache.spark.rdd.RDD[String] = file:////root/spark/core/test.txt MapPartitionsRDD[1] at textFile at <console>:24

words: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at flatMap at <console>:25

pairs: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[3] at map at <console>:26

wordCounts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[4] at reduceByKey at <console>:27

相關標籤/搜索

00'notberepresentedasql.timestimest

Microsoft Office

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。