To write a Spark application in Java, you need to add a dependency on Spark. Spark is available through Maven Central at:

groupId = org.apache.spark

artifactId = spark-core_2.10

version = 1.3.0

In addition, if you wish to access an HDFS cluster, you need to add a dependency on hadoop-client for your version of HDFS. Some common HDFS version tags are listed on the third party distributions page.

groupId = org.apache.hadoop

artifactId = hadoop-client

version = 
       

Finally, you need to import some Spark classes into your program. Add the following lines:

importorg.apache.spark.api.java.JavaSparkContext

importorg.apache.spark.api.java.JavaRDD

importorg.apache.spark.SparkConf

SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);

JavaSparkContext  sc  =  new JavaSparkContext(conf);

./bin/spark-shell –master local[4]  --jars code.jar

List<Integer>  data = Arrays.asList(1 ,2, 3, 4, 5);

JavaRDD<Integer> distData  =  sc.parallelize(data);

distData.reduce((a,b) -> a+b )

但需要注意，上面这个内部函数是利用了Java8的函数机制，在老版本的Java中没有该特性，所以需要利用org.apache.spark.api.java.function 来取代。

JavaRDD<String> distFile = sc.textFile(“data.txt”);

JavaRDD<String> lines = sc.textFile(“data.txt”);

JavaRDD<Integer> lineLengths = lines.map(line -> line.length);

Int totalLength = lineLengths.reduce((a,b) -> a + b);

before the reduce, which would cause lineLengths to be saved in memory after the first time it is computed.

// map function

Class GetLineLength implements Function<String, Integer>{

  Public Integer call(String s){

    Return s.length;

}

}

// reduce function

Class Sum implements Function2<Integer, Integer, Integer>{

  Public Integer call(Integer a, Integer b){

    Return a+b;

}

}

// driver function

JavaRDD<String> file = sc.textFile(“xxx”);

JavaRDD<Integer> lineLengths = file.map(new GetLineLength());

Int total = lineLengths.reduce(new Sum());

JavaRDD<String> file = sc.textFile(“xxx”);

JavaPairRDD<String, Integer> paris = file.mapToPair(line -> new Tuple2(line, 1));

JavaPairRDD<String , Integer> counters = pairs.reduceByKey((a,b) -> a+b);

Counters.sortByKey();

Counters.collect()可以将结果表示成Java的数组

Transformation

 解释

Map(func)

传递原数据集的每一个element，返回一个新的分布式数据集。映射

Filter(func)

从原数据集中选择出满足给定条件的elements，组成新的RDD

flatMap(func)

第一步和map一致，然后再进行扁平化，即把数组合并

MapPartitions(func)

也和map类似，只是这里操作的是分区block,不是每一个element,所以func(iterator<T>  =>  iterator<U>)  对block的数据进行运算

mapPartitionsWithIndex

(func)

类似mapPartitions，不过会传递一个分区号，所以输入变成

Func(  (Int, Iterator<U>)  =>  Iterator<U>)

Sample(withReplacement, 

fraction, seed)

Sample a fraction fraction of the data, with or without replacement, using a given random number generator seed.

union(otherDataset)

Return a new dataset that contains the union of the elements in the source dataset and the argument.

intersection(otherDataset)

Return a new RDD that contains the intersection of elements in the source dataset and the argument.

distinct([numTasks]))

Return a new dataset that contains the distinct elements of the source dataset.

groupByKey([numTasks])

When called on a dataset of (K, V) pairs, returns a dataset of (K, Iterable<V>) pairs. 

reduceByKey(func, [numTasks])

When called on a dataset of (K, V) pairs, returns a dataset of (K, V) pairs where the values for each key are aggregated using the given reduce function func, which must be of type (V,V) => V. Like in groupByKey, the number of reduce tasks is configurable through an optional second argument.

aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])

When called on a dataset of (K, V) pairs, returns a dataset of (K, U) pairs where the values for each key are aggregated using the given combine functions and a neutral "zero" value. Allows an aggregated value type that is different than the input value type, while avoiding unnecessary allocations. Like in groupByKey, the number of reduce tasks is configurable through an optional second argument.

sortByKey([ascending], [numTasks])

When called on a dataset of (K, V) pairs where K implements Ordered, returns a dataset of (K, V) pairs sorted by keys in ascending or descending order, as specified in the booleanascending argument.

join(otherDataset, [numTasks])

When called on datasets of type (K, V) and (K, W), returns a dataset of (K, (V, W)) pairs with all pairs of elements for each key. Outer joins are supported through leftOuterJoin,rightOuterJoin, and fullOuterJoin.

cogroup(otherDataset, [numTasks])

When called on datasets of type (K, V) and (K, W), returns a dataset of (K, (Iterable<V>, Iterable<W>)) tuples. This operation is also called groupWith.

cartesian(otherDataset)

When called on datasets of types T and U, returns a dataset of (T, U) pairs (all pairs of elements).

pipe(command, [envVars])

Pipe each partition of the RDD through a shell command, e.g. a Perl or bash script. RDD elements are written to the process's stdin and lines output to its stdout are returned as an RDD of strings.

coalesce(numPartitions)

Decrease the number of partitions in the RDD to numPartitions. Useful for running operations more efficiently after filtering down a large dataset.

repartition(numPartitions)

Reshuffle the data in the RDD randomly to create either more or fewer partitions and balance it across them. This always shuffles all data over the network.

repartitionAndSortWithinPartitions(partitioner)

Repartition the RDD according to the given partitioner and, within each resulting partition, sort records by their keys. This is more efficient than calling repartition and then sorting within each partition because it can push the sorting down into the shuffle machinery

Action

Meaning

reduce(func)

Aggregate the elements of the dataset using a function func (which takes two arguments and returns one). The function should be commutative and associative so that it can be computed correctly in parallel.

collect()

Return all the elements of the dataset as an array at the driver program. This is usually useful after a filter or other operation that returns a sufficiently small subset of the data.

count()

Return the number of elements in the dataset.

first()

Return the first element of the dataset (similar to take(1)).

take(n)

Return an array with the first n elements of the dataset.

takeSample(withReplacement,num, [seed])

Return an array with a random sample of num elements of the dataset, with or without replacement, optionally pre-specifying a random number generator seed.

takeOrdered(n, [ordering])

Return the first n elements of the RDD using either their natural order or a custom comparator.

saveAsTextFile(path)

Write the elements of the dataset as a text file (or set of text files) in a given directory in the local filesystem, HDFS or any other Hadoop-supported file system. Spark will call toString on each element to convert it to a line of text in the file.

saveAsSequenceFile(path) 

Write the elements of the dataset as a Hadoop SequenceFile in a given path in the local filesystem, HDFS or any other Hadoop-supported file system. This is available on RDDs of key-value pairs that either implement Hadoop's Writable interface. In Scala, it is also available on types that are implicitly convertible to Writable (Spark includes conversions for basic types like Int, Double, String, etc).

saveAsObjectFile(path) 

Write the elements of the dataset in a simple format using Java serialization, which can then be loaded usingSparkContext.objectFile().

countByKey()

Only available on RDDs of type (K, V). Returns a hashmap of (K, Int) pairs with the count of each key.

foreach(func)

Run a function func on each element of the dataset. This is usually done for side effects such as updating an accumulator variable (see below) or interacting with external storage systems.

Storage Level

Meaning

MEMORY_ONLY

Store RDD as deserialized Java objects in the JVM. If the RDD does not fit in memory, some partitions will not be cached and will be recomputed on the fly each time they're needed. This is the default level.

MEMORY_AND_DISK

Store RDD as deserialized Java objects in the JVM. If the RDD does not fit in memory, store the partitions that don't fit on disk, and read them from there when they're needed.

MEMORY_ONLY_SER

Store RDD as serialized Java objects (one byte array per partition). This is generally more space-efficient than deserialized objects, especially when using a fast serializer, but more CPU-intensive to read.

MEMORY_AND_DISK_SER

Similar to MEMORY_ONLY_SER, but spill partitions that don't fit in memory to disk instead of recomputing them on the fly each time they're needed.

DISK_ONLY

Store the RDD partitions only on disk.

MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc.

Same as the levels above, but replicate each partition on two cluster nodes.

OFF_HEAP (experimental)

Store RDD in serialized format in Tachyon. Compared to MEMORY_ONLY_SER, OFF_HEAP reduces garbage collection overhead and allows executors to be smaller and to share a pool of memory, making it attractive in environments with large heaps or multiple concurrent applications. Furthermore, as the RDDs reside in Tachyon, the crash of an executor does not lead to losing the in-memory cache. In this mode, the memory in Tachyon is discardable. Thus, Tachyon does not attempt to reconstruct a block that it evicts from memory.

Broadcast<int[]> broadcastVar = sc.broadcast(new int[]{1,2,3});

broadcastVar.value();  // return [1, 2, 3]

Accumulator<Integer> accum = sc.accumulator(0);

Sc.parallelize(Arrays.asList(1,2,3,4)).foreach(x->accum.add(x));

Accum.value();

Class VectorAccumulatorParam  implements  AccumulatorParam<Vector>{

  Public Vector zero(){

    Return Vector.zero(initialValue.size());

}

  Public Vector addInPlace(Vector v1， Vector v2){

    V1.addInPlace(v2); return v1; // 就是自定义两个对象怎么进行加法运算啦

}

}

这样，我们就可以自定义加法类型

Accumulator<Vector> vecAccum = sc.accumulator(new Vector(..), new VectorAccumulatorParam())

第一个参数就是初始值，后面就是我们定义的加法