翻译：Hadoop权威指南之Spark-2

发表于 | 更新于

Scala独立应用

在Spark shell中运行了一个小程序之后，你可能想要把它打包成自包含应用，这样就可以多次运行了。

示例19-1. 使用Spark找出最高气温的Scala应用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import org.apache.spark.SparkContext._
import org.apache.spark.{SparkConf, SparkContext}

object MaxTemperature {
    def main(args: Array[String]) {
        val conf = new SparkConf().setAppName("Max Temperature")
        val sc = new SparkContext(conf)
        sc.textFile(args(0))
          .map(_.split("\t"))
          .filter(rec => (rec(1) != "9999" && rec(2).matches("[01459]")))
          .map(rec => (rec(0).toInt, rec(1).toInt))
          .reduceByKey((a, b) => Math.max(a, b))
          .saveAsTextFile(args(1))
    }
}

运行独立程序时，没有shell为我们提供SparkContext，我们需要自己创建。我们用一个SparkConf来创建这个实例。SparkConf可以用来向应用中传递多个Spark属性，这里我们仅仅设置应用的名字。

还有一些别的微小变化。首先是我们使用命令行参数来指定输入和输出路径。另外还使用了方法链来避免为每一个RDD创建中间变量，这样程序更紧凑，如果需要的话，我们仍然可以在Scala IDE中查看每次转换操作的类型信息。

并非所有的Spark定义的转换操作都可用于RDD类本身。在本例中，reduceByKey()（仅仅在键值对的RDD上起作用）实际上定义在PairRDDFunctions类中，但我们能用下面的import来让Scala隐含地把RDD[(Int, Int)]转为PairRDDFunctions：

1
import org.apache.spark.SparkContext._

这个import不同于Spark使用的隐式转型函数，因此理所当然地值得包含在程序中。

这一次我们使用spark-submit来运行这个程序，把包含编译后的Scala程序的JAR包作为参数传入，接着传入命令行参数（输入和输出路径）：

1
2
3
4
5
% spark-submit --class MaxTemperature --master local \
spark-examples.jar input/ncdc/micro-tab/sample.txt output
% cat output/part-*
(1950,22)
(1949,111)

我们还指定了两个选项：–class 告诉Spark应用类的名字，–master 指定作业的运行方式，local值告诉Spark在本地机器的单个JVM中运行，在“执行器和集群管理器”一节我们将会学到一些在集群中运行的选项。接下来，我们看看怎样用Java语言来使用Spark。

Java示例

Spark是使用Scala实现的，Scala是基于JVM的语言，可以和Java完美集成。同样的例子用Java来表达，很直接也很啰嗦（使用Java 8的lambda表达式可以使这个版本更紧凑）。

示例19-2. 使用Spark找出最高气温的Java应用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
public class MaxTemperatureSpark {
    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println("Usage: MaxTemperatureSpark <input path> <output path>");
            System.exit(-1);
        }
        SparkConf conf = new SparkConf();
        JavaSparkContext sc = new JavaSparkContext("local", "MaxTemperatureSpark", conf);
        JavaRDD<String> lines = sc.textFile(args[0]);
        JavaRDD<String[]> records = lines.map(new Function<String, String[]>() {
            @Override public String[] call(String s) {
                return s.split("\t");
            }
        });
        JavaRDD<String[]> filtered = records.filter(new Function<String[], Boolean>() {
            @Override public Boolean call(String[] rec) {
                return rec[1] != "9999" && rec[2].matches("[01459]");
            }
        });
        JavaPairRDD<Integer, Integer> tuples = filtered.mapToPair(
            new PairFunction<String[], Integer, Integer>() {
                @Override public Tuple2<Integer, Integer> call(String[] rec) {
                    return new Tuple2<Integer, Integer>(
                        Integer.parseInt(rec[0]), Integer.parseInt(rec[1]));
                }
        });
        JavaPairRDD<Integer, Integer> maxTemps = tuples.reduceByKey(
            new Function2<Integer, Integer, Integer>() {
                @Override public Integer call(Integer i1, Integer i2) {
                    return Math.max(i1, i2);
                }
        });
        maxTemps.saveAsTextFile(args[1]);
    }
}

在Spark的Java API中，一个RDD由JavaRDD的实例表示，在键值对RDD的特殊情况下是JavaPairRDD 。这两个类都实现了JavaRDDLike接口，该接口中可以找到操作RDD的大多数方法。

运行这个程序和运行Scala版本一样，除了类名字是MaxTemperatureSpark 。

Python示例

Spark也支持Python语言，API叫做PySpark。由于Python语言有lambda表达式，例子程序非常接近Scala的版本。

示例19-3. 使用Spark找出最高气温的Python应用

1
2
3
4
5
6
7
8
9
10
form pyspark import SparkContext
import re, sys

sc = SparkContext("local", "Max Temperature")
sc.textFile(sys.argv[1]) \
  .map(lambda s: s.split("\t")) \
  .filter(lambda rec: (rec[1] != "9999" and re.match("[01459]", rec[2]))) \
  .map(lambda rec: (int(rec[0]), int(rec[1]))) \
  .reduceByKey(max) \
  .saveAsTextFile(sys.argv[2])

注意在reduceByKey()的转换操作中，我们可以使用Python语言内建的max函数。

需要留意的一点是，这个程序是用CPython写的，Spark会创建一个Python子进程来执行用户的Python代码（在启动程序 launcher 和在集群上运行用户任务的执行器 executor 上）。两个进程间使用socket通讯来传递RDD分区数据。

要运行这个程序，只需指定Python文件即可：

1
2
3
% spark-submit --master local \
  ch19-spark/src/main/python/MaxTemperature.py \
  input/ncdc/micro-tab/sample.txt output

还可以使用pyspark命令，以交互模式运行Spark和Python。