第10课：实战java开发spark程序

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

一、环境搭建

1）安装jdk并配置环境变量

网上有很多参考资料，此处省略。

2）Maven的安装和配置

解压apache-maven-3.1.1-bin.zip，并把解压后的文件夹下的apache-maven-3.1.1文件夹移动到D:\Java下，如果没有Java这个文件夹的话，请自行创建新建系统变量   MAVEN_HOME 变量值：D:\Java\apache-maven-3.1.1。编辑系统变量 Path 添加变量值：;%MAVEN_HOME%\bin。在mave 的目录中修改conf/settings.xml， 在 localRepository属性后添加<localRepository>D:/repository</localRepository>修改maven下载jar 的位置。

3）eclipse 中java 和maven 的配置

点击 window ->java ->Installed JREs ->add ->standard vm  ,点击next ，然后选择jdk 的安装路径点击finish即可。点击window ->Maven ->Installations ->add 在弹出页面选择mave 的安装路径，然后点击finish。然后在列表中选择我们自己刚添加的那个maven信息。然后点击window ->Maven ->User Setings   在 右侧的User Settings 点击browse 现在maven  conf目录下的setttings.xml（主要是修改maven下载依赖包存放的位置）。

二、创建maven项目

1）创建maven项目

点击file -->new->others ->maven project  点击next，选择maven-archetype-quickstart，点击next，group id 为 com.dt.spark，artifact id 为 sparkApps，然后点击finish。

2）修改jdk 和pom文件

创建maven项目后，默认的jdk 是1.5要改成我们前面安装好的jdk1.8 。在项目上右击build path ->configure build path 。在弹出页面点击Libraries，选中jre systemlibrary 。点击edit，在弹出框选择workspace default jre ，然后点击finish。 然后在点击ok。将pom文件修改为如下内容，然后等待eclipse 下载好maven依赖的jar包，并编译工程。编译好工程后有个错误提示，在此错误列上，右击选择quick fix ，在弹出页面点击finish即可。

1. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   
2. xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   
3. <modelVersion>4.0.0</modelVersion>
   
4. 
   
5.   <groupId>com.dt.spark</groupId>
   
6.   <artifactId>SparkApps</artifactId>
   
7.   <version>0.0.1-SNAPSHOT</version>
   
8.   <packaging>jar</packaging>
   
9. 
   
10.   <name>SparkApps</name>
    
11.   <url>http://maven.apache.org</url>
    
12. 
    
13.   <properties>
    
14.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    
15.   </properties>
    
16. 
    
17.   <dependencies>
    
18.     <dependency>
    
19.       <groupId>junit</groupId>
    
20.       <artifactId>junit</artifactId>
    
21.       <version>3.8.1</version>
    
22.       <scope>test</scope>
    
23.     </dependency>
    
24.     <dependency>
    
25.        <groupId>org.apache.spark</groupId>
    
26.        <artifactId>spark-core_2.10</artifactId>
    
27.        <version>1.6.0</version>
    
28.     </dependency>
    
29.     <dependency>
    
30.         <groupId>org.apache.spark</groupId>
    
31.         <artifactId>spark-sql_2.10</artifactId>
    
32.         <version>1.6.0</version>
    
33.     </dependency>
    
34.     <dependency>
    
35.       <groupId>org.apache.spark</groupId>
    
36.       <artifactId>spark-hive_2.10</artifactId>
    
37.       <version>1.6.0</version>
    
38.     </dependency>
    
39.         <dependency>
    
40.               <groupId>org.apache.spark</groupId>
    
41.               <artifactId>spark-streaming_2.10</artifactId>
    
42.               <version>1.6.0</version>
    
43.         </dependency>
    
44.         <dependency>
    
45.               <groupId>org.apache.hadoop</groupId>
    
46.               <artifactId>hadoop-client</artifactId>
    
47.               <version>2.6.0</version>
    
48.         </dependency>
    
49.         <dependency>
    
50.               <groupId>org.apache.spark</groupId>
    
51.               <artifactId>spark-streaming-kafka_2.10</artifactId>
    
52.               <version>1.6.0</version>
    
53.         </dependency>
    
54.         <dependency>
    
55.               <groupId>org.apache.spark</groupId>
    
56.               <artifactId>spark-graphx_2.10</artifactId>
    
57.               <version>1.6.0</version>
    
58.         </dependency>
    
59.   </dependencies>
    
60. 
    
61.   <build>
    
62.      <sourceDirectory>src/main/java</sourceDirectory>
    
63.      <testSourceDirectory>src/main/test</testSourceDirectory>
    
64. 
    
65.      <plugins>
    
66.            <plugin>
    
67.                <artifactId>maven-assembly-plugin</artifactId>
    
68.                <configuration>
    
69.                  <descriptorRefs>
    
70.                    <descriptorRef>jar-with-dependencies</descriptorRef>
    
71.                  </descriptorRefs>
    
72.                  <archive>
    
73.                     <manifest>
    
74.                         <maniClass></maniClass>
    
75.                     </manifest>
    
76.                  </archive>
    
77.                </configuration>
    
78.                <executions>
    
79.                    <execution>
    
80.                       <id>make-assembly</id>
    
81.                       <phase>package</phase>
    
82.                       <goals>
    
83.                          <goal>single</goal>
    
84.                       </goals>
    
85.                    </execution>
    
86.                </executions>
    
87.            </plugin>
    
88.            <plugin>
    
89.                <groupId>org.codehaus.mojo</groupId>
    
90.                <artifactId>exec-maven-plugin</artifactId>
    
91.                <version>1.3.1</version>
    
92.                <executions>
    
93.                   <execution>
    
94.                      <goals>
    
95.                         <goal>exec</goal>
    
96.                      </goals>
    
97.                   </execution>
    
98.                </executions>
    
99.                <configuration>
    
100.                    <executable>java</executable>
     
101.                    <includeProjectDependencies>false</includeProjectDependencies>
     
102.                    <classpathScope>compile</classpathScope>
     
103.                    <mainClass>com.dt.spark.SparkApps.WordCount</mainClass>
     
104.                </configuration>
     
105.            </plugin>
     
106.            <plugin>
     
107.               <groupId>org.apache.maven.plugins</groupId>
     
108.               <artifactId>maven-compiler-plugin</artifactId>
     
109.               <configuration>
     
110.                   <source>1.6</source>
     
111.                   <target>1.6</target>
     
112.               </configuration>
     
113.            </plugin>
     
114.      </plugins>
     
115. </build>
     
116. 
     
117. </project>

复制代码

3）创建包路径以及java代码

在包路径com.dt.spark.SparkApps 上 右击 new ->package 在弹出页面name中填写com.dt.spark.SparkApps.cores,点击finish的。在包路径下com.dt.spark.SparkApps.cores上 右击 new ->class ，在弹出窗口中name 中填写 WordCount，点击finish。然后在WordCount 中编写如下代码。

1. import java.util.Arrays;
   
2. import org.apache.spark.SparkConf;
   
3. import org.apache.spark.api.java.JavaPairRDD;
   
4. import org.apache.spark.api.java.JavaRDD;
   
5. import org.apache.spark.api.java.JavaSparkContext;
   
6. import org.apache.spark.api.java.function.FlatMapFunction;
   
7. import org.apache.spark.api.java.function.Function2;
   
8. import org.apache.spark.api.java.function.PairFunction;
   
9. import org.apache.spark.api.java.function.VoidFunction;
   
10. import scala.Function;
    
11. import scala.Tuple2;
    
12. 
    
13. /**
    
14. * 使用java的方式开发进行本地测试spark的 wordcount 程序
    
15. * @author DT大数据梦工厂
    
16. *
    
17. */
    
18. public classWordCount {
    
19. 
    
20.     public static void main(String[] args) {
    
21. 
    
22.          /**
    
23.            * 第1步：创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息，
    
24.            * 例如说通过setMaster来设置程序要链接的Spark集群的Master的URL,如果设置
    
25.            * 为local，则代表Spark程序在本地运行，特别适合于机器配置条件非常差（例如
    
26.            * 只有1G的内存）的初学者       *
    
27.            */
    
28.         SparkConfconf=newSparkConf().setAppName("Spark WordCountwritten by java").setMaster("local");
    
29. 
    
30.         /**
    
31.            * 第2步：创建SparkContext对象
    
32.            * SparkContext是Spark程序所有功能的唯一入口，无论是采用Scala、Java、Python、R等都必须有一个SparkContext(不同的语言具体的类名称不同，如果是java 的为javaSparkContext)
    
33.            * SparkContext核心作用：初始化Spark应用程序运行所需要的核心组件，包括DAGScheduler、TaskScheduler、SchedulerBackend
    
34.            * 同时还会负责Spark程序往Master注册程序等
    
35.            * SparkContext是整个Spark应用程序中最为至关重要的一个对象
    
36.            */
    
37.         JavaSparkContextsc=new JavaSparkContext(conf); //其底层就是scala的sparkcontext
    
38. 
    
39. 
    
40.           /**
    
41.            * 第3步：根据具体的数据来源（HDFS、HBase、Local FS、DB、S3等）通过SparkContext来创建RDD
    
42.            * JavaRDD的创建基本有三种方式：根据外部的数据来源（例如HDFS）、根据Scala集合、由其它的RDD操作
    
43.            * 数据会被JavaRDD划分成为一系列的Partitions，分配到每个Partition的数据属于一个Task的处理范畴
    
44.            */
    
45.         JavaRDD<String>lines=sc.textFile("D://spark-1.6.0-bin-hadoop2.6//README.md");
    
46. 
    
47.          /**
    
48.            * 第4步：对初始的JavaRDD进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
    
49.            *     第4.1步：讲每一行的字符串拆分成单个的单词
    
50.            */
    
51.         JavaRDD<String>words= lines.flatMap(new FlatMapFunction<String,String>(){//如果是scala由于Sam转化所以可以写成一行代码
    
52. 
    
53.             @Override
    
54.             public Iterable<String>call(String line)throwsException {
    
55.                 // TODO Auto-generated method stub
    
56. 
    
57.                 return Arrays.asList(line.split(" "));
    
58.             }
    
59. 
    
60.         });
    
61.         /**
    
62.            * 第4步：对初始的JavaRDD进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
    
63.            *     第4.2步：在单词拆分的基础上对每个单词实例计数为1，也就是word => (word, 1)
    
64.            */
    
65.         JavaPairRDD<String,Integer>pairs=words.mapToPair(new PairFunction<String,String, Integer>() {
    
66. 
    
67.             @Override
    
68.             public Tuple2<String,Integer> call(String word) throwsException {
    
69.                 // TODO Auto-generated method stub
    
70.                 return new Tuple2<String,Integer>(word,1);
    
71.             }
    
72.         });
    
73. 
    
74. 
    
75.         /**
    
76.            * 第4步：对初始的RDD进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
    
77.            *     第4.3步：在每个单词实例计数为1基础之上统计每个单词在文件中出现的总次数
    
78.            */
    
79.         JavaPairRDD<String,Integer>wordsCount=pairs.reduceByKey(new Function2<Integer,Integer, Integer>() { //对相同的Key，进行Value的累计（包括Local和Reducer级别同时Reduce）
    
80. 
    
81.             @Override
    
82.             public Integer call(Integer v1, Integer v2) throws Exception {
    
83.                 // TODO Auto-generated method stub
    
84.                 return v1+v2;
    
85.             }
    
86.         });
    
87. 
    
88.         wordsCount.foreach(new VoidFunction<Tuple2<String,Integer>>(){
    
89. 
    
90.             @Override
    
91.             public void call(Tuple2<String,Integer> pairs)throwsException {
    
92.                 // TODO Auto-generated method stub
    
93.                 System.out.println(pairs._1+" : " +pairs._2);
    
94.             }
    
95.         });
    
96. 
    
97.         sc.close();
    
98. 
    
99.     }
    
100. 
     
101. }

复制代码

在代码区右击run as -> java application 。来运行此程序并查看运行结果。如果要开发cluster的代码，请参考前面第8课或者第9课。

注：本学习笔记来自DT大数据梦工厂

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏1 回帖

关键词：java开发 Spark Park SPAR Java Spark scala DT_Spark 大数据

正在学习中，感谢楼主！！！

lzjsltt 发表于 2016-4-26 22:43
正在学习中，感谢楼主！！！