Spark编程指南入门之Java篇一-基本知识

1. Spark的Java开发包

Spark提供Java的开发包，当前最新版本是2.0.2版本：spark-core_2.11-2.0.2.jar，可以从下面链接下载：
http://central.maven.org/maven2/org/apache/spark/spark-core_2.11/2.0.2/spark-core_2.11-2.0.2.jar
或者通过Maven配置：

[xml] view
plain copy

<dependency>  

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

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

    <version>2.0.2</version>  

</dependency>  

* Spark 2.0.2版本需要Java 7或以上，本文使用Java 1.8.0_72版本

2. 初始化Spark

要使用Spark，第一步必须创建JavaSparkContext对象：

[java] view
plain copy

import org.apache.spark.SparkConf;  

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

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

  

public class HelloSpark {  

      

    public static void main(String[] args) {  

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

        try (JavaSparkContext jsc = new JavaSparkContext(conf)) {  

            // do something here  

        }  

    }  

  

}  

上述代码通过SparkConf创建JavaSparkContext，SparkConf默认去读取Spark.*的配置文件，也可以通过调用set的方法配置属性，例如上述的setMaster和setAppName。通过set方法配置的属性会覆盖读取的配置文件属性，SparkConf里面的所有set方法都支持链式调用chaining，例如上述的setMaster("local").setAppName("HelloSpark")。

* 上述代码使用了Java 7的try-with-resources语句，用于自动关闭JavaSparkContext
* setAppName：设置应用名字，此名字会在Spark web UI显示
* setMaster：设置主节点URL，本例使用“local”是指本机单线程，另外还有以下选项：

local[K]：本机K线程
local[*]：本机多线程，线程数与服务器核数相同
spark://HOST:PORT：Spark集群地址和端口，默认端口为7077
mesos://HOST:PORT：Mesos集群地址和端口，默认端口为5050
yarn：YARN集群

详细请参考以下链接：
http://spark.apache.org/docs/latest/submitting-applications.html#master-urls
3. 弹性分布式数据集Resilient Distributed Datasets(RDDs)

Spark使用了一个叫弹性分布式数据集resilient distributed dataset(RDD)，它是支持并行操作的容错性元素集合。RDD可以通过2种方式创建：

- 并行化一个存在的集合
- 引用外部存储系统(例如共享文件系统、HDFS、HBase或者任何支持Hadoop InputFormat的数据源)的数据集

3.1 并行集合Parallelized Collections

并行集合Parallelized Collections可以通过调用JavaSparkContext的parallelize方法创建，例如：

[java] view
plain copy

import java.util.Arrays;  

import java.util.List;  

  

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

// jsc是上述代码已经创建的JavaSparkContext实例  

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

上述JavaRDD一旦被创建，它就可以被并行操作，例如调用reduce方法计算集合里面所有元素的总和：

[java] view
plain copy

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

* 上述代码使用Java 8的lambda语法(这里不做介绍)，在Java 8之前，上述代码可以使用下面代码实现：

[java] view
plain copy

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

  

distData.reduce(new Function2<Integer, Integer, Integer>() {  

              

    private static final long serialVersionUID = 1L;  

  

    @Override  

    public Integer call(Integer a, Integer b) throws Exception {  

        return a + b;  

    }  

      

});  

并行集合的一个重要参数是用于分布数据集的分区数量，Spark会对每一个数据分区执行一个任务，通常地，每一个CPU可以处理2到4个分区。一般地，Spark会基于集群的资源情况自动设定分区数量，不过也可以使用parallelize的第二个参数自定义，例如jsc.parallelize(data,
5)，这样的话，数据就会被分为5个分区，Spark可以并行处理这5个分区，提高处理效率。

3.2 外部数据集External Datasets

上面提过，Spark可以从任何支持Hadoop InputFormat的数据源创建分布式数据集，包括本地文件系统，HDFS，Cassandra, HBase, Amazon S3等等，Spark支持文本文件，序列化文件和其它Hadoop InputFormat。例如，以下代码通过调用SparkContext的textFile方法读取本地文件系统创建一个文本格式的RDD：

[java] view
plain copy

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

* HDFS的路径是以hdfs://开头，Amazon S3的路径是以s3://开头

3.3 Spark读取文件的一些注意事项

- 如果路径是本地的文件系统路径，那么这个路径必须能够被在所有工作节点的同样路径访问，你可以把文件复制到所有工作节点的对应路径或者使用共享文件系统

- 所有基于文件输入的Spark方法，包括textFile，路径可以是目录，压缩文件和通配符。例如，textFile("/my/directory"), textFile("/my/directory/*.txt"), and textFile("/my/directory/*.gz")

- textFile方法的第二个参数是可选的，用于控制文件的分区数量。Spark默认对文件的每一个block都创建一个分区(HDFS的默认block大小是64MB)，通过设置该参数可以修改文件的分区数量，但是分区数量不能少于block的数量