我们首先提出这样一个简单的需求：

现在要分析某网站的访问日志信息，统计来自不同IP的用户访问的次数，从而通过Geo信息来获得来访用户所在国家地区分布状况。这里我拿我网站的日志记录行示例，如下所示：

121.205.198.92 - - [21/Feb/2014:00:00:07 +0800] "GET /archives/417.html HTTP/1.1" 200 11465 "http://shiyanjun.cn/archives/417.html/" "Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko/20100101 Firefox/11.0"121.205.198.92 - - [21/Feb/2014:00:00:11 +0800] "POST /wp-comments-post.php HTTP/1.1" 302 26 "http://shiyanjun.cn/archives/417.html/" "Mozilla/5.0 (Windows NT 5.1; rv:23.0) Gecko/20100101 Firefox/23.0"121.205.198.92 - - [21/Feb/2014:00:00:12 +0800] "GET /archives/417.html/ HTTP/1.1" 301 26 "http://shiyanjun.cn/archives/417.html/" "Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko/20100101 Firefox/11.0"121.205.198.92 - - [21/Feb/2014:00:00:12 +0800] "GET /archives/417.html HTTP/1.1" 200 11465 "http://shiyanjun.cn/archives/417.html" "Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko/20100101 Firefox/11.0"121.205.241.229 - - [21/Feb/2014:00:00:13 +0800] "GET /archives/526.html HTTP/1.1" 200 12080 "http://shiyanjun.cn/archives/526.html/" "Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko/20100101 Firefox/11.0"121.205.241.229 - - [21/Feb/2014:00:00:15 +0800] "POST /wp-comments-post.php HTTP/1.1" 302 26 "http://shiyanjun.cn/archives/526.html/" "Mozilla/5.0 (Windows NT 5.1; rv:23.0) Gecko/20100101 Firefox/23.0"

Java实现Spark应用程序（Application）

我们实现的统计分析程序，有如下几个功能点：

从HDFS读取日志数据文件

将每行的第一个字段（IP地址）抽取出来

统计每个IP地址出现的次数

根据每个IP地址出现的次数进行一个降序排序

根据IP地址，调用GeoIP库获取IP所属国家

打印输出结果，每行的格式：[国家代码] IP地址 频率

下面，看我们使用Java实现的统计分析应用程序代码，如下所示：

package org.shirdrn.spark.job;import java.io.File;import java.io.IOException;import java.util.Arrays;import java.util.Collections;import java.util.Comparator;import java.util.List;import java.util.regex.Pattern;import org.apache.commons.logging.Log;import org.apache.commons.logging.LogFactory;import org.apache.spark.api.java.JavaPairRDD;import org.apache.spark.api.java.JavaRDD;import org.apache.spark.api.java.JavaSparkContext;import org.apache.spark.api.java.function.FlatMapFunction;import org.apache.spark.api.java.function.Function2;import org.apache.spark.api.java.function.PairFunction;import org.shirdrn.spark.job.maxmind.Country;import org.shirdrn.spark.job.maxmind.LookupService;import scala.Serializable;import scala.Tuple2;public class IPAddressStats implements Serializable { private static final long serialVersionUID = 8533489548835413763L; private static final Log LOG = LogFactory.getLog(IPAddressStats.class); private static final Pattern SPACE = Pattern.compile(" "); private transient LookupService lookupService; private transient final String geoIPFile; public IPAddressStats(String geoIPFile) { this.geoIPFile = geoIPFile; try { // lookupService: get country code from a IP address File file = new File(this.geoIPFile); LOG.info("GeoIP file: " + file.getAbsolutePath()); lookupService = new AdvancedLookupService(file, LookupService.GEOIP_MEMORY_CACHE); } catch (IOException e) { throw new RuntimeException(e); } } @SuppressWarnings("serial") public void stat(String[] args) { JavaSparkContext ctx = new JavaSparkContext(args[0], "IPAddressStats", System.getenv("SPARK_HOME"), JavaSparkContext.jarOfClass(IPAddressStats.class)); JavaRDD<String> lines = ctx.textFile(args[1], 1); // splits and extracts ip address filed JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override public Iterable<String> call(String s) { // 121.205.198.92 - - [21/Feb/2014:00:00:07 +0800] "GET /archives/417.html HTTP/1.1" 200 11465 "http://shiyanjun.cn/archives/417.html/" "Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko/20100101 Firefox/11.0" // ip address return Arrays.asList(SPACE.split(s)[0]); } }); // map JavaPairRDD<String, Integer> ones = words.map(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String s) { return new Tuple2<String, Integer>(s, 1); } }); // reduce JavaPairRDD<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() { @Override public Integer call(Integer i1, Integer i2) { return i1 + i2; } }); List<Tuple2<String, Integer>> output = counts.collect(); // sort statistics result by value Collections.sort(output, new Comparator<Tuple2<String, Integer>>() { @Override public int compare(Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) { if(t1._2 < t2._2) { return 1; } else if(t1._2 > t2._2) { return -1; } return 0; } }); writeTo(args, output); } private void writeTo(String[] args, List<Tuple2<String, Integer>> output) { for (Tuple2<?, ?> tuple : output) { Country country = lookupService.getCountry((String) tuple._1); LOG.info("[" + country.getCode() + "] " + tuple._1 + "\t" + tuple._2); } } public static void main(String[] args) { // ./bin/run-my-java-example org.shirdrn.spark.job.IPAddressStats spark://m1:7077 hdfs://m1:9000/user/shirdrn/pleted ResultTask(0, 0)14/03/10 22:17:32 INFO scheduler.DAGScheduler: Stage 0 (collect at IPAddressStats.java:77) finished in 0.314 s14/03/10 22:17:32 INFO scheduler.TaskSchedulerImpl: Remove TaskSet 0.0 from pool14/03/10 22:17:32 INFO spark.SparkContext: Job finished: collect at IPAddressStats.java:77, took 4.870958309 s14/03/10 22:17:32 INFO job.IPAddressStats: [CN] 58.246.49.218 31214/03/10 22:17:32 INFO job.IPAddressStats: [KR] 1.234.83.77 30014/03/10 22:17:32 INFO job.IPAddressStats: [CN] 120.43.11.16 21214/03/10 22:17:32 INFO job.IPAddressStats: [CN] 110.85.72.254 20714/03/10 22:17:32 INFO job.IPAddressStats: [CN] 27.150.229.134 18514/03/10 22:17:32 INFO job.IPAddressStats: [HK] 180.178.52.181 18114/03/10 22:17:32 INFO job.IPAddressStats: [CN] 120.37.210.212 18014/03/10 22:17:32 INFO job.IPAddressStats: [CN] 222.77.226.83 17614/03/10 22:17:32 INFO job.IPAddressStats: [CN] 120.43.11.205 16914/03/10 22:17:32 INFO job.IPAddressStats: [CN] 120.43.9.19 165...

我们也可以通过Web控制台来查看当前执行应用程序（Application）的状态信息，通过Master节点的8080端口（如：http://m1:8080/）就能看到集群的应用程序（Application）状态信息。

另外，需要说明的时候，如果在Unix环境下使用Eclipse使用Java开发Spark应用程序，也能够直接通过Eclipse连接Spark集群，并提交开发的应用程序，然后交给集群去处理。

总结

以上就是本文关于详解Java编写并运行spark应用程序的方法的全部内容，希望对大家有所帮助。有什么问题可以随时留言，小编会及时回复大家。