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 SparkContextimport 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。
作者:qinm08
链接:https://www.jianshu.com/p/77d652359a30