详解Java编写并运行spark应用程序的方法

103 浏览量更新于2023-03-03 收藏 81KB PDF 举报

身份认证购VIP最低享 7 折!

领优惠券(最高得80元）

资源详情

资源推荐

详解详解Java编写并运行编写并运行spark应用程序的方法应用程序的方法

主要介绍了详解Java编写并运行spark应用程序的方法，内容详细，结合了作者实际工作中的问题进行具体分析，具有一定参考价值。

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

现在要分析某网站的访问日志信息，统计来自不同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 + " " + 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/wwwlog20140222.log /home/shirdrn/cloud/programs/spark-0.9.0-incubating-bin-hadoop1/java-examples/GeoIP_DATABASE.dat

if (args.length < 3) {

System.err.println("Usage: IPAddressStats <master> <inFile> <GeoIPFile>");

System.err.println(" Example: org.shirdrn.spark.job.IPAddressStats spark://m1:7077 hdfs://m1:9000/user/shirdrn/wwwlog20140222.log /home/shirdrn/cloud/programs/spark-0.9.0-incubating-bin-hadoop1/java-examples/GeoIP_DATABASE.dat");

System.exit(1);

}

本内容试读结束，登录后可阅读更多

下载后可阅读完整内容，剩余3页未读，立即下载