需求分析和技术架构广告点击系统实时分析
广告来自于广告或者移动App等，广告需要设定在具体的广告位，当用户点击广告的时候，一般都会通过ajax或Socket往后台发送日志数据，在这里我们是要做基于SparkStreaming做实时在线统计。那么数据就需要放进消息系统（Kafka）中，我们的Spark Streaming应用程序就会去Kafka中Pull数据过来进行计算和消费，并把计算后的数据放入到持久化系统中（MySQL）
广告点击系统实时分析的意义：因为可以在线实时的看见广告的投放效果，就为广告的更大规模的投入和调整打下了坚实的基础，从而为公司带来最大化的经济回报。
核心需求：
1、实时黑名单动态过滤出有效的用户广告点击行为：因为黑名单用户可能随时出现，所以需要动态更新；
2、在线计算广告点击流量；
3、Top3热门广告；
4、每个广告流量趋势；
5、广告点击用户的区域分布分析
6、最近一分钟的广告点击量；
7、整个广告点击Spark Streaming处理程序7*24小时运行； 数据格式：
时间、用户、广告、城市等 技术细节：
在线计算用户点击的次数分析，屏蔽IP等；
使用updateStateByKey或者mapWithState进行不同地区广告点击排名的计算；
Spark Streaming+Spark SQL+Spark Core等综合分析数据；
使用Window类型的操作；
高可用和性能调优等等；
流量趋势，一般会结合DB等；
Spark Core 
/**
*
*/
package com.tom.spark.SparkApps.sparkstreaming;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
/**
* 数据生成代码，Kafka Producer产生数据
*/
public class MockAdClickedStat {
/**
* @param args
*/
public static void main(String[] args) {
final Random random = new Random();
final String[] provinces = new String[]{"Guangdong", "Zhejiang", "Jiangsu", "Fujian"};
final Map<String, String[]> cities = new HashMap<String, String[]>();
cities.put("Guangdong", new String[]{"Guangzhou", "Shenzhen", "Dongguan"});
cities.put("Zhejiang", new String[]{"Hangzhou", "Wenzhou", "Ningbo"});
cities.put("Jiangsu", new String[]{"Nanjing", "Suzhou", "Wuxi"});
cities.put("Fujian", new String[]{"Fuzhou", "Xiamen", "Sanming"});
final String[] ips = new String[] {
"192.168.112.240",
"192.168.112.239",
"192.168.112.245",
"192.168.112.246",
"192.168.112.247",
"192.168.112.248",
"192.168.112.249",
"192.168.112.250",
"192.168.112.251",
"192.168.112.252",
"192.168.112.253",
"192.168.112.254",
};
/**
* Kafka相关的基本配置信息
*/
Properties kafkaConf = new Properties();
kafkaConf.put("serializer.class", "kafka.serializer.StringEncoder");
kafkaConf.put("metadeta.broker.list", "Master:9092,Worker1:9092,Worker2:9092");
ProducerConfig producerConfig = new ProducerConfig(kafkaConf);
final Producer<Integer, String> producer = new Producer<Integer, String>(producerConfig);
new Thread(new Runnable() {
public void run() {
while(true) {
//在线处理广告点击流的基本数据格式：timestamp、ip、userID、adID、province、city
Long timestamp = new Date().getTime();
String ip = ips[random.nextInt(12)]; //可以采用网络上免费提供的ip库
int userID = random.nextInt(10000);
int adID = random.nextInt(100);
String province = provinces[random.nextInt(4)];
String city = cities.get(province)[random.nextInt(3)];
String clickedAd = timestamp + "\t" + ip + "\t" + userID + "\t" + adID + "\t" + province + "\t" + city;
producer.send(new KeyedMessage<Integer, String>("AdClicked", clickedAd));
try {
Thread.sleep(50);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}).start();
}
}
package com.tom.spark.SparkApps.sparkstreaming;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import kafka.serializer.StringDecoder;
import org.apache.spark.SparkConf;
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.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.api.java.JavaStreamingContextFactory;
import org.apache.spark.streaming.kafka.KafkaUtils;
import com.google.common.base.Optional;
import scala.Tuple2;
/**
* 数据处理，Kafka消费者
*/
public class AdClickedStreamingStats {
/**
* @param args
*/
public static void main(String[] args) {
// TODO Auto-generated method stub
//好处：1、checkpoint 2、工厂
final SparkConf conf = new SparkConf().setAppName("SparkStreamingOnKafkaDirect").setMaster("hdfs://Master:7077/");
final String checkpointDirectory = "hdfs://Master:9000/library/SparkStreaming/CheckPoint_Data";
JavaStreamingContextFactory factory = new JavaStreamingContextFactory() {
public JavaStreamingContext create() {
// TODO Auto-generated method stub
return createContext(checkpointDirectory, conf);
}
};
/**
* 可以从失败中恢复Driver，不过还需要指定Driver这个进程运行在Cluster，并且在提交应用程序的时候制定--supervise;
*/
JavaStreamingContext javassc = JavaStreamingContext.getOrCreate(checkpointDirectory, factory);
/**
* 第三步：创建Spark Streaming输入数据来源input Stream:
* 1、数据输入来源可以基于File、HDFS、Flume、Kafka、Socket等
* 2、在这里我们指定数据来源于网络Socket端口，Spark Streaming连接上该端口并在运行的时候一直监听该端口的数据
* (当然该端口服务首先必须存在），并且在后续会根据业务需要不断有数据产生（当然对于Spark Streaming
* 应用程序的运行而言，有无数据其处理流程都是一样的）
* 3、如果经常在每间隔5秒钟没有数据的话不断启动空的Job其实会造成调度资源的浪费，因为并没有数据需要发生计算；所以
* 实际的企业级生成环境的代码在具体提交Job前会判断是否有数据，如果没有的话就不再提交Job；
*/
//创建Kafka元数据来让Spark Streaming这个Kafka Consumer利用
Map<String, String> kafkaParameters = new HashMap<String, String>();
kafkaParameters.put("metadata.broker.list", "Master:9092,Worker1:9092,Worker2:9092");
Set<String> topics = new HashSet<String>();
topics.add("SparkStreamingDirected");
JavaPairInputDStream<String, String> adClickedStreaming = KafkaUtils.createDirectStream(javassc,
String.class, String.class,
StringDecoder.class, StringDecoder.class,
kafkaParameters,
topics);
/**因为要对黑名单进行过滤，而数据是在RDD中的，所以必然使用transform这个函数；
* 但是在这里我们必须使用transformToPair，原因是读取进来的Kafka的数据是Pair<String,String>类型,
* 另一个原因是过滤后的数据要进行进一步处理，所以必须是读进的Kafka数据的原始类型
*
* 在此再次说明，每个Batch Duration中实际上讲输入的数据就是被一个且仅被一个RDD封装的，你可以有多个
* InputDStream，但其实在产生job的时候，这些不同的InputDStream在Batch Duration中就相当于Spark基于HDFS
* 数据操作的不同文件来源而已罢了。
*/
JavaPairDStream<String, String> filteredadClickedStreaming = adClickedStreaming.transformToPair(new Function<JavaPairRDD<String,String>, JavaPairRDD<String,String>>() {
public JavaPairRDD<String, String> call(
JavaPairRDD<String, String> rdd) throws Exception {
/**
* 在线黑名单过滤思路步骤：
* 1、从数据库中获取黑名单转换成RDD，即新的RDD实例封装黑名单数据；
* 2、然后把代表黑名单的RDD的实例和Batch Duration产生的RDD进行Join操作，
* 准确的说是进行leftOuterJoin操作，也就是说使用Batch Duration产生的RDD和代表黑名单的RDD实例进行
* leftOuterJoin操作，如果两者都有内容的话，就会是true，否则的话就是false
*
* 我们要留下的是leftOuterJoin结果为false；
*
*/
final List<String> blackListNames = new ArrayList<String>();
JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();
jdbcWrapper.doQuery("SELECT * FROM blacklisttable", null, new ExecuteCallBack() {
public void resultCallBack(ResultSet result) throws Exception {
while(result.next()){
blackListNames.add(result.getString(1));
}
}
});
List<Tuple2<String, Boolean>> blackListTuple = new ArrayList<Tuple2<String,Boolean>>();
for(String name : blackListNames) {
blackListTuple.add(new Tuple2<String, Boolean>(name, true));
}
List<Tuple2<String, Boolean>> blacklistFromListDB = blackListTuple; //数据来自于查询的黑名单表并且映射成为<String, Boolean>
JavaSparkContext jsc = new JavaSparkContext(rdd.context());
/**
* 黑名单的表中只有userID，但是如果要进行join操作的话就必须是Key-Value，所以在这里我们需要
* 基于数据表中的数据产生Key-Value类型的数据集合
*/
JavaPairRDD<String, Boolean> blackListRDD = jsc.parallelizePairs(blacklistFromListDB);
/**
* 进行操作的时候肯定是基于userID进行join，所以必须把传入的rdd进行mapToPair操作转化成为符合格式的RDD
*
*/
JavaPairRDD<String, Tuple2<String, String>> rdd2Pair = rdd.mapToPair(new PairFunction<Tuple2<String,String>, String, Tuple2<String, String>>() {
public Tuple2<String, Tuple2<String, String>> call(
Tuple2<String, String> t) throws Exception {
// TODO Auto-generated method stub
String userID = t._2.split("\t")[2];
return new Tuple2<String, Tuple2<String,String>>(userID, t);
}
});
JavaPairRDD<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> joined = rdd2Pair.leftOuterJoin(blackListRDD);
JavaPairRDD<String, String> result = joined.filter(new Function<Tuple2<String,Tuple2<Tuple2<String,String>,Optional<Boolean>>>, Boolean>() {
public Boolean call(
Tuple2<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> tuple)
throws Exception {
// TODO Auto-generated method stub
Optional<Boolean> optional = tuple._2._2;
if(optional.isPresent() && optional.get()){
return false;
} else {
return true;
}
}
}).mapToPair(new PairFunction<Tuple2<String,Tuple2<Tuple2<String,String>,Optional<Boolean>>>, String, String>() {
public Tuple2<String, String> call(
Tuple2<String, Tuple2<Tuple2<String, String>, Optional<Boolean>>> t)
throws Exception {
// TODO Auto-generated method stub
return t._2._1;
}
});
return result;
}
});
//广告点击的基本数据格式：timestamp、ip、userID、adID、province、city
JavaPairDStream<String, Long> pairs = filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {
public Tuple2<String, Long> call(Tuple2<String, String> t) throws Exception {
String[] splited=t._2.split("\t");
String timestamp = splited[0]; //YYYY-MM-DD
String ip = splited[1];
String userID = splited[2];
String adID = splited[3];
String province = splited[4];
String city = splited[5];
String clickedRecord = timestamp + "_" +ip + "_"+userID+"_"+adID+"_"
+province +"_"+city;
return new Tuple2<String, Long>(clickedRecord, 1L);
}
});
/**
* 第4.3步：在单词实例计数为1基础上，统计每个单词在文件中出现的总次数
*/
JavaPairDStream<String, Long> adClickedUsers= pairs.reduceByKey(new Function2<Long, Long, Long>() {
public Long call(Long i1, Long i2) throws Exception{
return i1 + i2;
}
});
/*判断有效的点击，复杂化的采用机器学习训练模型进行在线过滤 简单的根据ip判断1天不超过100次；也可以通过一个batch duration的点击次数
判断是否非法广告点击，通过一个batch来判断是不完整的，还需要一天的数据也可以每一个小时来判断。*/
JavaPairDStream<String, Long> filterClickedBatch = adClickedUsers.filter(new Function<Tuple2<String,Long>, Boolean>() {
public Boolean call(Tuple2<String, Long> v1) throws Exception {
if (1 < v1._2){
//更新一些黑名单的数据库表
return false;
} else {
return true;
}
}
});
//filterClickedBatch.print();
//写入数据库
filterClickedBatch.foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {
public Void call(JavaPairRDD<String, Long> rdd) throws Exception {
rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {
public void call(Iterator<Tuple2<String, Long>> partition) throws Exception {
//使用数据库连接池的高效读写数据库的方式将数据写入数据库mysql
//例如一次插入 1000条 records，使用insertBatch 或 updateBatch
//插入的用户数据信息：userID,adID,clickedCount,time
//这里面有一个问题，可能出现两条记录的key是一样的，此时需要更新累加操作
List<UserAdClicked> userAdClickedList = new ArrayList<UserAdClicked>();
while(partition.hasNext()) {
Tuple2<String, Long> record = partition.next();
String[] splited = record._1.split("\t");
UserAdClicked userClicked = new UserAdClicked();
userClicked.setTimestamp(splited[0]);
userClicked.setIp(splited[1]);
userClicked.setUserID(splited[2]);
userClicked.setAdID(splited[3]);
userClicked.setProvince(splited[4]);
userClicked.setCity(splited[5]);
userAdClickedList.add(userClicked);
}
final List<UserAdClicked> inserting = new ArrayList<UserAdClicked>();
final List<UserAdClicked> updating = new ArrayList<UserAdClicked>();
JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();
//表的字段timestamp、ip、userID、adID、province、city、clickedCount
for(final UserAdClicked clicked : userAdClickedList) {
jdbcWrapper.doQuery("SELECT clickedCount FROM adclicked WHERE"
+ " timestamp =? AND userID = ? AND adID = ?",
new Object[]{clicked.getTimestamp(), clicked.getUserID(),
clicked.getAdID()}, new ExecuteCallBack() {
public void resultCallBack(ResultSet result) throws Exception {
// TODO Auto-generated method stub
if(result.next()) {
long count = result.getLong(1);
clicked.setClickedCount(count);
updating.add(clicked);
} else {
inserting.add(clicked);
clicked.setClickedCount(1L);
}
}
});
}
//表的字段timestamp、ip、userID、adID、province、city、clickedCount
List<Object[]> insertParametersList = new ArrayList<Object[]>();
for(UserAdClicked insertRecord : inserting) {
insertParametersList.add(new Object[] {
insertRecord.getTimestamp(),
insertRecord.getIp(),
insertRecord.getUserID(),
insertRecord.getAdID(),
insertRecord.getProvince(),
insertRecord.getCity(),
insertRecord.getClickedCount()
});
}
jdbcWrapper.doBatch("INSERT INTO adclicked VALUES(?, ?, ?, ?, ?, ?, ?)", insertParametersList);
//表的字段timestamp、ip、userID、adID、province、city、clickedCount
List<Object[]> updateParametersList = new ArrayList<Object[]>();
for(UserAdClicked updateRecord : updating) {
updateParametersList.add(new Object[] {
updateRecord.getTimestamp(),
updateRecord.getIp(),
updateRecord.getUserID(),
updateRecord.getAdID(),
updateRecord.getProvince(),
updateRecord.getCity(),
updateRecord.getClickedCount() + 1
});
}
jdbcWrapper.doBatch("UPDATE adclicked SET clickedCount = ? WHERE"
+ " timestamp =? AND ip = ? AND userID = ? AND adID = ? "
+ "AND province = ? AND city = ?", updateParametersList);
}
});
return null;
}
});
//再次过滤，从数据库中读取数据过滤黑名单
JavaPairDStream<String, Long> blackListBasedOnHistory = filterClickedBatch.filter(new Function<Tuple2<String,Long>, Boolean>() {
public Boolean call(Tuple2<String, Long> v1) throws Exception {
//广告点击的基本数据格式：timestamp,ip,userID,adID,province,city
String[] splited = v1._1.split("\t"); //提取key值
String date =splited[0];
String userID =splited[2];
String adID =splited[3];
//查询一下数据库同一个用户同一个广告id点击量超过50次列入黑名单
//接下来 根据date、userID、adID条件去查询用户点击广告的数据表，获得总的点击次数
//这个时候基于点击次数判断是否属于黑名单点击
int clickedCountTotalToday = 81 ;
if (clickedCountTotalToday > 50) {
return true;
}else {
return false ;
}
}
});
//map操作，找出用户的id
JavaDStream<String> blackListuserIDBasedInBatchOnhistroy =blackListBasedOnHistory.map(new Function<Tuple2<String,Long>, String>() {
public String call(Tuple2<String, Long> v1) throws Exception {
// TODO Auto-generated method stub
return v1._1.split("\t")[2];
}
});
//有一个问题，数据可能重复，在一个partition里面重复，这个好办；
//但多个partition不能保证一个用户重复，需要对黑名单的整个rdd进行去重操作。
//rdd去重了，partition也就去重了，一石二鸟，一箭双雕// 找出了黑名单，下一步就写入黑名单数据库表中
JavaDStream<String> blackListUniqueuserBasedInBatchOnhistroy = blackListuserIDBasedInBatchOnhistroy.transform(new Function<JavaRDD<String>, JavaRDD<String>>() {
public JavaRDD<String> call(JavaRDD<String> rdd) throws Exception {
// TODO Auto-generated method stub
return rdd.distinct();
}
});
// 下一步写入到数据表中
blackListUniqueuserBasedInBatchOnhistroy.foreachRDD(new Function<JavaRDD<String>, Void>() {
public Void call(JavaRDD<String> rdd) throws Exception {
rdd.foreachPartition(new VoidFunction<Iterator<String>>() {
public void call(Iterator<String> t) throws Exception {
// TODO Auto-generated method stub
//插入的用户信息可以只包含：useID
//此时直接插入黑名单数据表即可。
//写入数据库
List<Object[]> blackList = new ArrayList<Object[]>();
while(t.hasNext()) {
blackList.add(new Object[]{t.next()});
}
JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();
jdbcWrapper.doBatch("INSERT INTO blacklisttable values (?)", blackList);
}
});
return null;
}
});
/**广告点击累计动态更新,每个updateStateByKey都会在Batch Duration的时间间隔的基础上进行广告点击次数的更新，
* 更新之后我们一般都会持久化到外部存储设备上，在这里我们存储到MySQL数据库中
*/
JavaPairDStream<String, Long> updateStateByKeyDSteam = filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {
public Tuple2<String, Long> call(Tuple2<String, String> t)
throws Exception {
String[] splited=t._2.split("\t");
String timestamp = splited[0]; //YYYY-MM-DD
String ip = splited[1];
String userID = splited[2];
String adID = splited[3];
String province = splited[4];
String city = splited[5];
String clickedRecord = timestamp + "_" +ip + "_"+userID+"_"+adID+"_"
+province +"_"+city;
return new Tuple2<String, Long>(clickedRecord, 1L);
}
}).updateStateByKey(new Function2<List<Long>, Optional<Long>, Optional<Long>>() {
public Optional<Long> call(List<Long> v1, Optional<Long> v2)
throws Exception {
// v1:当前的Key在当前的Batch Duration中出现的次数的集合，例如{1，1，1，。。。，1}
// v2:当前的Key在以前的Batch Duration中积累下来的结果；
Long clickedTotalHistory = 0L;
if(v2.isPresent()){
clickedTotalHistory = v2.get();
}
for(Long one : v1) {
clickedTotalHistory += one;
}
return Optional.of(clickedTotalHistory);
}
});
updateStateByKeyDSteam.foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {
public Void call(JavaPairRDD<String, Long> rdd) throws Exception {
rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {
public void call(Iterator<Tuple2<String, Long>> partition) throws Exception {
//使用数据库连接池的高效读写数据库的方式将数据写入数据库mysql
//例如一次插入 1000条 records，使用insertBatch 或 updateBatch
//插入的用户数据信息：timestamp、adID、province、city
//这里面有一个问题，可能出现两条记录的key是一样的，此时需要更新累加操作
List<AdClicked> AdClickedList = new ArrayList<AdClicked>();
while(partition.hasNext()) {
Tuple2<String, Long> record = partition.next();
String[] splited = record._1.split("\t");
AdClicked adClicked = new AdClicked();
adClicked.setTimestamp(splited[0]);
adClicked.setAdID(splited[1]);
adClicked.setProvince(splited[2]);
adClicked.setCity(splited[3]);
adClicked.setClickedCount(record._2);
AdClickedList.add(adClicked);
}
final List<AdClicked> inserting = new ArrayList<AdClicked>();
final List<AdClicked> updating = new ArrayList<AdClicked>();
JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();
//表的字段timestamp、ip、userID、adID、province、city、clickedCount
for(final AdClicked clicked : AdClickedList) {
jdbcWrapper.doQuery("SELECT clickedCount FROM adclickedcount WHERE"
+ " timestamp = ? AND adID = ? AND province = ? AND city = ?",
new Object[]{clicked.getTimestamp(), clicked.getAdID(),
clicked.getProvince(), clicked.getCity()}, new ExecuteCallBack() {
public void resultCallBack(ResultSet result) throws Exception {
// TODO Auto-generated method stub
if(result.next()) {
long count = result.getLong(1);
clicked.setClickedCount(count);
updating.add(clicked);
} else {
inserting.add(clicked);
clicked.setClickedCount(1L);
}
}
});
}
//表的字段timestamp、ip、userID、adID、province、city、clickedCount
List<Object[]> insertParametersList = new ArrayList<Object[]>();
for(AdClicked insertRecord : inserting) {
insertParametersList.add(new Object[] {
insertRecord.getTimestamp(),
insertRecord.getAdID(),
insertRecord.getProvince(),
insertRecord.getCity(),
insertRecord.getClickedCount()
});
}
jdbcWrapper.doBatch("INSERT INTO adclickedcount VALUES(?, ?, ?, ?, ?)", insertParametersList);
//表的字段timestamp、ip、userID、adID、province、city、clickedCount
List<Object[]> updateParametersList = new ArrayList<Object[]>();
for(AdClicked updateRecord : updating) {
updateParametersList.add(new Object[] {
updateRecord.getClickedCount(),
updateRecord.getTimestamp(),
updateRecord.getAdID(),
updateRecord.getProvince(),
updateRecord.getCity()
});
}
jdbcWrapper.doBatch("UPDATE adclickedcount SET clickedCount = ? WHERE"
+ " timestamp =? AND adID = ? AND province = ? AND city = ?", updateParametersList);
}
});
return null;
}
});
/**
* 对广告点击进行TopN计算，计算出每天每个省份Top5排名的广告
* 因为我们直接对RDD进行操作，所以使用了transfomr算子；
*/
updateStateByKeyDSteam.transform(new Function<JavaPairRDD<String,Long>, JavaRDD<Row>>() {
public JavaRDD<Row> call(JavaPairRDD<String, Long> rdd) throws Exception {
JavaRDD<Row> rowRDD = rdd.mapToPair(new PairFunction<Tuple2<String,Long>, String, Long>() {
public Tuple2<String, Long> call(Tuple2<String, Long> t)
throws Exception {
// TODO Auto-generated method stub
String[] splited=t._1.split("_");
String timestamp = splited[0]; //YYYY-MM-DD
String adID = splited[3];
String province = splited[4];
String clickedRecord = timestamp + "_" + adID + "_" + province;
return new Tuple2<String, Long>(clickedRecord, t._2);
}
}).reduceByKey(new Function2<Long, Long, Long>() {
public Long call(Long v1, Long v2) throws Exception {
// TODO Auto-generated method stub
return v1 + v2;
}
}).map(new Function<Tuple2<String,Long>, Row>() {
public Row call(Tuple2<String, Long> v1) throws Exception {
// TODO Auto-generated method stub
String[] splited=v1._1.split("_");
String timestamp = splited[0]; //YYYY-MM-DD
String adID = splited[3];
String province = splited[4];
return RowFactory.create(timestamp, adID, province, v1._2);
}
});
StructType structType = DataTypes.createStructType(Arrays.asList(
DataTypes.createStructField("timestamp", DataTypes.StringType, true),
DataTypes.createStructField("adID", DataTypes.StringType, true),
DataTypes.createStructField("province", DataTypes.StringType, true),
DataTypes.createStructField("clickedCount", DataTypes.LongType, true)
));
HiveContext hiveContext = new HiveContext(rdd.context());
DataFrame df = hiveContext.createDataFrame(rowRDD, structType);
df.registerTempTable("topNTableSource");
DataFrame result = hiveContext.sql("SELECT timestamp, adID, province, clickedCount, FROM"
+ " (SELECT timestamp, adID, province,clickedCount, "
+ "ROW_NUMBER() OVER(PARTITION BY province ORDER BY clickeCount DESC) rank "
+ "FROM topNTableSource) subquery "
+ "WHERE rank <= 5");
return result.toJavaRDD();
}
}).foreachRDD(new Function<JavaRDD<Row>, Void>() {
public Void call(JavaRDD<Row> rdd) throws Exception {
// TODO Auto-generated method stub
rdd.foreachPartition(new VoidFunction<Iterator<Row>>() {
public void call(Iterator<Row> t) throws Exception {
// TODO Auto-generated method stub
List<AdProvinceTopN> adProvinceTopN = new ArrayList<AdProvinceTopN>();
while(t.hasNext()) {
Row row = t.next();
AdProvinceTopN item = new AdProvinceTopN();
item.setTimestamp(row.getString(0));
item.setAdID(row.getString(1));
item.setProvince(row.getString(2));
item.setClickedCount(row.getLong(3));
adProvinceTopN.add(item);
}
// final List<AdProvinceTopN> inserting = new ArrayList<AdProvinceTopN>();
// final List<AdProvinceTopN> updating = new ArrayList<AdProvinceTopN>();
JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();
Set<String> set = new HashSet<String>();
for(AdProvinceTopN item: adProvinceTopN){
set.add(item.getTimestamp() + "_" + item.getProvince());
}
//表的字段timestamp、adID、province、clickedCount
ArrayList<Object[]> deleteParametersList = new ArrayList<Object[]>();
for(String deleteRecord : set) {
String[] splited = deleteRecord.split("_");
deleteParametersList.add(new Object[]{
splited[0],
splited[1]
});
}
jdbcWrapper.doBatch("DELETE FROM adprovincetopn WHERE timestamp = ? AND province = ?", deleteParametersList);
//表的字段timestamp、ip、userID、adID、province、city、clickedCount
List<Object[]> insertParametersList = new ArrayList<Object[]>();
for(AdProvinceTopN insertRecord : adProvinceTopN) {
insertParametersList.add(new Object[] {
insertRecord.getClickedCount(),
insertRecord.getTimestamp(),
insertRecord.getAdID(),
insertRecord.getProvince()
});
}
jdbcWrapper.doBatch("INSERT INTO adprovincetopn VALUES (?, ?, ?, ?)", insertParametersList);
}
});
return null;
}
});
/**
* 计算过去半个小时内广告点击的趋势
* 广告点击的基本数据格式：timestamp、ip、userID、adID、province、city
*/
filteredadClickedStreaming.mapToPair(new PairFunction<Tuple2<String,String>, String, Long>() {
public Tuple2<String, Long> call(Tuple2<String, String> t)
throws Exception {
String splited[] = t._2.split("\t");
String adID = splited[3];
String time = splited[0]; //Todo:后续需要重构代码实现时间戳和分钟的转换提取。此处需要提取出该广告的点击分钟单位
return new Tuple2<String, Long>(time + "_" + adID, 1L);
}
}).reduceByKeyAndWindow(new Function2<Long, Long, Long>() {
public Long call(Long v1, Long v2) throws Exception {
// TODO Auto-generated method stub
return v1 + v2;
}
}, new Function2<Long, Long, Long>() {
public Long call(Long v1, Long v2) throws Exception {
// TODO Auto-generated method stub
return v1 - v2;
}
}, Durations.minutes(30), Durations.milliseconds(5)).foreachRDD(new Function<JavaPairRDD<String,Long>, Void>() {
public Void call(JavaPairRDD<String, Long> rdd) throws Exception {
// TODO Auto-generated method stub
rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Long>>>() {
public void call(Iterator<Tuple2<String, Long>> partition)
throws Exception {
List<AdTrendStat> adTrend = new ArrayList<AdTrendStat>();
// TODO Auto-generated method stub
while(partition.hasNext()) {
Tuple2<String, Long> record = partition.next();
String[] splited = record._1.split("_");
String time = splited[0];
String adID = splited[1];
Long clickedCount = record._2;
/**
* 在插入数据到数据库的时候具体需要哪些字段？time、adID、clickedCount;
* 而我们通过J2EE技术进行趋势绘图的时候肯定是需要年、月、日、时、分这个维度的，所以我们在这里需要
* 年月日、小时、分钟这些时间维度；
*/
AdTrendStat adTrendStat = new AdTrendStat();
adTrendStat.setAdID(adID);
adTrendStat.setClickedCount(clickedCount);
adTrendStat.set_date(time); //Todo:获取年月日
adTrendStat.set_hour(time); //Todo:获取小时
adTrendStat.set_minute(time);//Todo:获取分钟
adTrend.add(adTrendStat);
}
final List<AdTrendStat> inserting = new ArrayList<AdTrendStat>();
final List<AdTrendStat> updating = new ArrayList<AdTrendStat>();
JDBCWrapper jdbcWrapper = JDBCWrapper.getJDBCInstance();
//表的字段timestamp、ip、userID、adID、province、city、clickedCount
for(final AdTrendStat trend : adTrend) {
final AdTrendCountHistory adTrendhistory = new AdTrendCountHistory();
jdbcWrapper.doQuery("SELECT clickedCount FROM adclickedtrend WHERE"
+ " date =? AND hour = ? AND minute = ? AND AdID = ?",
new Object[]{trend.get_date(), trend.get_hour(), trend.get_minute(),
trend.getAdID()}, new ExecuteCallBack() {
public void resultCallBack(ResultSet result) throws Exception {
// TODO Auto-generated method stub
if(result.next()) {
long count = result.getLong(1);
adTrendhistory.setClickedCountHistoryLong(count);
updating.add(trend);
} else {
inserting.add(trend);
}
}
});
}
//表的字段date、hour、minute、adID、clickedCount
List<Object[]> insertParametersList = new ArrayList<Object[]>();
for(AdTrendStat insertRecord : inserting) {
insertParametersList.add(new Object[] {
insertRecord.get_date(),
insertRecord.get_hour(),
insertRecord.get_minute(),
insertRecord.getAdID(),
insertRecord.getClickedCount()
});
}
jdbcWrapper.doBatch("INSERT INTO adclickedtrend VALUES(?, ?, ?, ?, ?)", insertParametersList);
//表的字段date、hour、minute、adID、clickedCount
List<Object[]> updateParametersList = new ArrayList<Object[]>();
for(AdTrendStat updateRecord : updating) {
updateParametersList.add(new Object[] {
updateRecord.getClickedCount(),
updateRecord.get_date(),
updateRecord.get_hour(),
updateRecord.get_minute(),
updateRecord.getAdID()
});
}
jdbcWrapper.doBatch("UPDATE adclickedtrend SET clickedCount = ? WHERE"
+ " date =? AND hour = ? AND minute = ? AND AdID = ?"
, updateParametersList);
}
});
return null;
}
});;
/**
* Spark Streaming 执行引擎也就是Driver开始运行，Driver启动的时候是位于一条新的线程中的，当然其内部有消息循环体，用于
* 接收应用程序本身或者Executor中的消息，
*/
javassc.start();
javassc.awaitTermination();
javassc.close();
}
private static JavaStreamingContext createContext(String checkpointDirectory, SparkConf conf) {
// If you do not see this printed, that means the StreamingContext has been loaded
// from the new checkpoint
System.out.println("Creating new context");
// Create the context with a 5 second batch size
JavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(10));
ssc.checkpoint(checkpointDirectory);
return ssc;
}
}
class JDBCWrapper {
private static JDBCWrapper jdbcInstance = null;
private static LinkedBlockingQueue<Connection> dbConnectionPool = new LinkedBlockingQueue<Connection>();
static {
try {
Class.forName("com.mysql.jdbc.Driver");
} catch (ClassNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
public static JDBCWrapper getJDBCInstance() {
if(jdbcInstance == null) {
synchronized (JDBCWrapper.class) {
if(jdbcInstance == null) {
jdbcInstance = new JDBCWrapper();
}
}
}
return jdbcInstance;
}
private JDBCWrapper() {
for(int i = 0; i < 10; i++){
try {
Connection conn = DriverManager.getConnection("jdbc:mysql://Master:3306/sparkstreaming","root", "root");
dbConnectionPool.put(conn);
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
public synchronized Connection getConnection() {
while(0 == dbConnectionPool.size()){
try {
Thread.sleep(20);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
return dbConnectionPool.poll();
}
public int[] doBatch(String sqlText, List<Object[]> paramsList){
Connection conn = getConnection();
PreparedStatement preparedStatement = null;
int[] result = null;
try {
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement(sqlText);
for(Object[] parameters: paramsList) {
for(int i = 0; i < parameters.length; i++){
preparedStatement.setObject(i + 1, parameters[i]);
}
preparedStatement.addBatch();
}
result = preparedStatement.executeBatch();
conn.commit();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
if(preparedStatement != null) {
try {
preparedStatement.close();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
if(conn != null) {
try {
dbConnectionPool.put(conn);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
return result;
}
public void doQuery(String sqlText, Object[] paramsList, ExecuteCallBack callback){
Connection conn = getConnection();
PreparedStatement preparedStatement = null;
ResultSet result = null;
try {
preparedStatement = conn.prepareStatement(sqlText);
for(int i = 0; i < paramsList.length; i++){
preparedStatement.setObject(i + 1, paramsList[i]);
}
result = preparedStatement.executeQuery();
try {
callback.resultCallBack(result);
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
if(preparedStatement != null) {
try {
preparedStatement.close();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
if(conn != null) {
try {
dbConnectionPool.put(conn);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
}
interface ExecuteCallBack {
void resultCallBack(ResultSet result) throws Exception;
}
class UserAdClicked {
private String timestamp;
private String ip;
private String userID;
private String adID;
private String province;
private String city;
private Long clickedCount;
public String getTimestamp() {
return timestamp;
}
public void setTimestamp(String timestamp) {
this.timestamp = timestamp;
}
public String getIp() {
return ip;
}
public void setIp(String ip) {
this.ip = ip;
}
public String getUserID() {
return userID;
}
public void setUserID(String userID) {
this.userID = userID;
}
public String getAdID() {
return adID;
}
public void setAdID(String adID) {
this.adID = adID;
}
public String getProvince() {
return province;
}
public void setProvince(String province) {
this.province = province;
}
public String getCity() {
return city;
}
public void setCity(String city) {
this.city = city;
}
public Long getClickedCount() {
return clickedCount;
}
public void setClickedCount(Long clickedCount) {
this.clickedCount = clickedCount;
}
}
class AdClicked {
private String timestamp;
private String adID;
private String province;
private String city;
private Long clickedCount;
public String getTimestamp() {
return timestamp;
}
public void setTimestamp(String timestamp) {
this.timestamp = timestamp;
}
public String getAdID() {
return adID;
}
public void setAdID(String adID) {
this.adID = adID;
}
public String getProvince() {
return province;
}
public void setProvince(String province) {
this.province = province;
}
public String getCity() {
return city;
}
public void setCity(String city) {
this.city = city;
}
public Long getClickedCount() {
return clickedCount;
}
public void setClickedCount(Long clickedCount) {
this.clickedCount = clickedCount;
}
}
class AdProvinceTopN {
private String timestamp;
private String adID;
private String province;
private Long clickedCount;
public String getTimestamp() {
return timestamp;
}
public void setTimestamp(String timestamp) {
this.timestamp = timestamp;
}
public String getAdID() {
return adID;
}
public void setAdID(String adID) {
this.adID = adID;
}
public String getProvince() {
return province;
}
public void setProvince(String province) {
this.province = province;
}
public Long getClickedCount() {
return clickedCount;
}
public void setClickedCount(Long clickedCount) {
this.clickedCount = clickedCount;
}
}
class AdTrendStat {
private String _date;
private String _hour;
private String _minute;
private String adID;
private Long clickedCount;
public String get_date() {
return _date;
}
public void set_date(String _date) {
this._date = _date;
}
public String get_hour() {
return _hour;
}
public void set_hour(String _hour) {
this._hour = _hour;
}
public String get_minute() {
return _minute;
}
public void set_minute(String _minute) {
this._minute = _minute;
}
public String getAdID() {
return adID;
}
public void setAdID(String adID) {
this.adID = adID;
}
public Long getClickedCount() {
return clickedCount;
}
public void setClickedCount(Long clickedCount) {
this.clickedCount = clickedCount;
}
}
class AdTrendCountHistory{
private Long clickedCountHistoryLong;
public Long getClickedCountHistoryLong() {
return clickedCountHistoryLong;
}
public void setClickedCountHistoryLong(Long clickedCountHistoryLong) {
this.clickedCountHistoryLong = clickedCountHistoryLong;
}
}
|
