3000门徒内部训练绝密视频（泄密版）第10课：彻底实战详解使用Java开发Spark程序

彻底实战详解使用Java开发Spark程序

第一步：下载Eclipse IDE for JAVA Developer

第二步：解压并启动Eclipse

第三步：创建Maven工程

第四步：使用maven-archetype-quickstart，设定一些包名

第五步：通过BuildPath把默认的J2EE 1.5变成Java1.8

第六步：配置pom.xml，添加程序开发时的相关依赖，并配置具体build打包的信息

POM.xml

有各种依赖的支持

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.10</artifactId>
<version>1.6.2</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.10</artifactId>
<version>1.6.2</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.10</artifactId>
<version>1.6.2</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.10</artifactId>
<version>1.6.2</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka_2.10</artifactId>
<version>1.6.2</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-graphx_2.10</artifactId>
<version>1.6.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.6.0</version>
</dependency>

http://maven.outofmemory.cn/org.apache.spark

package com.tom.spark.SparkApps.cores;

import java.util.Arrays;

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;

/**
* 使用Java的方式开发本地测试Spark的WordCount程序
* @author
*
*/
public class WordCount {

public static void main(String[] args) {
// TODO Auto-generated method stub
/**
* 第一步：创建Spark的配置对象，SparkConf，设置Spark程序的运行时的配置信息
* 例如通过setMaster来设置程序要连接的spark集群的Master的URL，如果设置
* 为local，则代表Spark程序在本地运行，特别适合于机器配置条件非常差的初学者
*
*/
SparkConf conf = new SparkConf().setAppName("Spark WordCount written by Java").setMaster("local");
/**
* 第二步：创建SparkContext对象
* SparkContext是Spark程序所有功能的唯一入口，无论采用Scala、Java、Python、R等都必须有一个SparkContext(不同的语言具体的类名称不同，Java则为JavaSparkContext)
* SparkContext核心作用：初始化Spark应用程序运行所需要的核心组件，包括DAGScheduler、TaskScheduler、SchedulerBackEnd
* 同时还会负责Spark程序往Master注册程序等
* SparkContext是整个Spark应用程序中最为至关重要的一个对象
*/
JavaSparkContext sc = new JavaSparkContext(conf);
/**
* 第三步：根据具体的数据来源（HDFS、HBase、Local FS、DB、S3等）通过SparkContext来创建JavaRDD
* JavaRDD的创建基本有三种方式：根据外部的数据来源（例如HDFS）、根据Scala集合、由其他JavaRDD操作
* 数据会被JavaRDD划分成为一系列的Partitions，分配到每个Partition的数据属于一个Task的处理范畴
*/
JavaRDD<String> lines = sc.textFile("F:/channel.txt",1);
/**
* 第四步：对初始的JavaRDD进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
* 第4.1步：将每一行的字符串拆分成单个的单词
*/
JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>(){ //如果shiScala，由于SAM转换，所以可以写成val words = lines.flatMap(_.split(" "))

@Override
public Iterable<String> call(String line) throws Exception {
return Arrays.asList(line.split(" "));
}
});

/**
* 第4.2步：在单词拆分的基础上对每个单词实例计数为1，也就是word => (word, 1)
*/
JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {

@Override
public Tuple2<String, Integer> call(String word) throws Exception {
// TODO Auto-generated method stub
return new Tuple2<String, Integer> (word, 1);
}

});
/**
* 第4.3步：在单词实例计数为1基础上，统计每个单词在文件中出现的总次数
*/
JavaPairRDD<String, Integer> wordsCount = pairs.reduceByKey(new Function2<Integer, Integer, Integer>(){
//对相同的key，进行Value的累加（包括Local和Reducer级别同时Reduce）
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
// TODO Auto-generated method stub
return v1 + v2;
}

});

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

@Override
public void call(Tuple2<String, Integer> pairs) throws Exception {
// TODO Auto-generated method stub
System.out.println(pairs._1 + " : " + pairs._2);
}
});

sc.close();
}

}

作业：放在集群上跑