A Scala Standalone Application
在Spark shell中运行了一个小程序之后，你可能想要把它打包成自包含应用，这样就可以多次运行了。
示例19-1. 使用Spark找出最高气温的Scala应用
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中查看每次转变（transformation）的类型信息。
并非所有的Spark定义的transformation都可用于RDD类本身。在本例中，reduceByKey()（仅仅在键值对的RDD上起作用）实际上定义在PairRDDFunctions类中，但我们能用下面的import来让Scala隐含地把RDD[(Int, Int)]转为PairRDDFunctions：
import org.apache.spark.SparkContext._
这个import不同于Spark使用的隐式转型函数，因此理所当然地值得包含在程序中。
这一次我们使用spark-submit来运行这个程序，把包含编译后的Scala程序的JAR包作为参数传入，接着传入命令行参数（输入输出路径）：
% 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 指定job的运行方式，local值告诉Spark在本地机器的单个JVM中运行，在“Executors and Cluster Managers”一节我们将会学到在集群中运行的选项。接下来，我们看看怎样用Java语言来使用Spark。
A Java Example
Spark是使用Scala实现的，Scala是基于JVM的语言，可以和Java完美集成。同样的例子用Java来表达，很直接，也很啰嗦（使用Java 8的lambda表达式可以使这个版本更紧凑）。
示例19-2. 使用Spark找出最高气温的Java应用
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 。
A Python Example
Spark也支持Python语言，API叫做PySpark。由于Python语言有lambda表达式，例子程序非常接近Scala的版本。
示例19-3. 使用Spark找出最高气温的Python应用
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文件即可：
% spark-submit --master local \
ch19-spark/src/main/python/MaxTemperature.py \
input/ncdc/micro-tab/sample.txt output
还可以使用pyspark命令，以交互模式运行Spark和Python。
