Java開発Sparkプログラム
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">
<modelVersion>4.0.0</modelVersion>
<groupId>com.dt.spark</groupId>
<artifactId>SparkApps</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>SparkApps</name>
<url>http://maven.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-graphx_2.10</artifactId>
<version>1.6.0</version>
</dependency>
</dependencies>
<build>
<sourceDirectory>src/main/java</sourceDirectory>
<testSourceDirectory>src/main/test</testSourceDirectory>
<plugins>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<archive>
<manifest>
<mainClass></mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>exec-maven-plugin</artifactId>
<version>1.2.1</version>
<executions>
<execution>
<goals>
<goal>exec</goal>
</goals>
</execution>
</executions>
<configuration>
<executable>java</executable>
<includeProjectDependencies>true</includeProjectDependencies>
<includePluginDependencies>false</includePluginDependencies>
<classpathScope>compile</classpathScope>
<mainClass>com.dt.spark.App</mainClass>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.6</source>
<target>1.6</target>
</configuration>
</plugin>
</plugins>
</build>
</project>
/** * ：WordCount.java * * ：Sundujing - [email protected] * * ：2016 5 9 9:53:28 * * ：Sundujing */
package com.dt.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;
import scala.Tuple2;
/** * [ ： ] * * @author Sundujing - [email protected] * @version 1.0 Created on 2016 5 9 9:53:28 */
public class WordCount {
public static void main(String[] args)
{
/** * ， Spark SparkConf， Spark ， * setMaster Spark Master URL， * local， Spark ， */
SparkConf conf=new SparkConf().setAppName("Spark workcount written by Java").setMaster("local");
/** * ， SparkContext * SparkContext Spark ， Scala，java，python，R * SparkContext（ ， Java ， JavaSparkContext） * Spark Master * SparkContext Spark */
JavaSparkContext sc =new JavaSparkContext(conf);// Scala SparkContext
/** * ， （HDFS，HBase，Local，FS，DB，S3 ）， JavaSparkContext JavaRDD * JavaRDD ： （ HDFS）， * Scala ， RDD RDD Partition， * Partition Task */
JavaRDD<String> lines = sc.textFile("D://spark-1.6.1-bin-hadoop2.6//README.md");
JavaRDD words = lines.flatMap(new FlatMapFunction<String,String>() {// Scala， SAM ， val words=lines.flatMap{line =>line.split(" ")}
public Iterable call(String line) throws Exception {
return Arrays.asList(line.split(" "));
}
});
/** * 4 ： JavaRDD Transformation ， map，filter ， * 4.1 ： 1， word =>(word,1) */
JavaPairRDD<String,Integer> pairs=words.mapToPair(new PairFunction<String,String,Integer>()
{
public Tuple2<String,Integer> call(String word) throws Exception{
return new Tuple2<String,Integer>(word,1);
}
});
/** * */
JavaPairRDD<String,Integer> wordCount=pairs.reduceByKey(new Function2<Integer,Integer,Integer>()
{
public Integer call(Integer v1,Integer v2)throws Exception
{
return v1+v2;
}
});
wordCount.foreach(new VoidFunction<Tuple2<String,Integer>>(){
public void call(Tuple2<String,Integer> pairs) throws Exception {
System.out.println(pairs._1()+":"+pairs._2());
}
});
sc.close();
}
}
クラスタモードコードは以下の通りです.
/** * ：WordCount_Cluster.java * * ：Sundujing - [email protected] * * ：2016 5 9 12:40:24 * * ：Sundujing */
package com.dt.spark.SparkApps.cores;
/** * [ ： ] * * @author Sundujing - [email protected] * @version 1.0 Created on 2016 5 9 12:40:24 */
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;
import java.util.Arrays;
import java.util.List;
public class WordCount_Cluster {
public static void main(String[] args)
{
SparkConf conf = new SparkConf();
//SparkConf conf = new SparkConf().setMaster("spark://yarn-client:7077").setAppName("WordCount by java");
//SparkConf conf = new SparkConf().setMaster("spark://172.171.51.131:7077").setAppName("WordCount by java");
//SparkConf conf = new SparkConf().set("spark.driver.host", "node1") .set("spark.driver.port", "60959").setAppName("WordCount by java");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaRDD lines = sc.textFile("hdfs://node1:8020/tmp/harryport.txt");
//JavaRDD<String> lines = sc.textFile("D://spark-1.6.1-bin-hadoop2.6//README.md");
JavaRDD words = lines.flatMap(new FlatMapFunction<String,String>() {
public Iterable call(String line) throws Exception {
return Arrays.asList(line.split(" "));
}
});
JavaPairRDD<String,Integer> paris = words.mapToPair(new PairFunction<String,String, Integer>() {
public Tuple2<String,Integer> call(String word) throws Exception {
return new Tuple2<String,Integer>(word, 1);
}
});
JavaPairRDD<String,Integer> wordCount = paris.reduceByKey(new Function2<Integer, Integer, Integer>() {
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
List<Tuple2<String,Integer>> list = wordCount.collect();
for(Tuple2<String,Integer> pari :list)
{
System.out.println(pari._1() + ":" + pari._2());
}
sc.close();
}
}1.パッケージExport-」jar file jar名javacount.jar 2.実行スクリプトを書くjavacount.sh
./spark-submit --class com.dt.spark.SparkApps.cores.WordCount_Cluster /root/javacount.jar
4.spark/binディレクトリ下
5.スクリプトを実行する. sh /root/javacount.sh