基于Spark Streaming对新闻网站项目案例分析

数据源  -> kafka  ->  spark streaming

date: 日期，yyyy-MM-dd格式 
timestamp: 时间戳 
userid: 用户id 
pageid: 页面id 
section: 版块 
action: 用户行为，两类，点击页面和注册

2020-12-20 1608451521565 364 422 fashion view
2020-12-20 1608451521565 38682 708 aviation view
2020-12-20 1608451521565 65444 270 internet view
2020-12-20 1608451521565 4805 250 tv-show view
2020-12-20 1608451521565 1130 743 movie view
2020-12-20 1608451521565 85320 605 carton view
2020-12-20 1608451521565 null 581 movie view
2020-12-20 1608451521565 null null null register

bin/kafka-console-consumer.sh --bootstrap-server bigdata-pro-m04:9092 --topic spark

package com.kfk.spark.news_analysis_project;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import java.util.Random;
/**
 * 访问日志Kafka Producer
 * @author : 蔡政洁
 * @email :caizhengjie888@icloud.com
 * @date : 2020/12/12
 * @time : 7:51 下午
 */
public class AccessProducer extends Thread{
    private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
    private static String date;
    // 版块内容
    private static String[] sections = new String[] {"country", "international", "sport",
            "entertainment", "movie", "carton",
            "tv-show", "technology", "internet",
            "car", "military", "funny",
            "fashion", "aviation", "government"};
    private static Random random = new Random();
    private static int[] newOldUserArr = new int[]{1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
    private Producer<String,String> producer;
    private String topic;
    /**
     * 构造函数
     * @param topic
     */
    public AccessProducer(String topic){
        this.topic = topic;
        producer = new KafkaProducer<String, String>(createProducerConfig());
        date = simpleDateFormat.format(new Date());
    }
    /**
     * createProducerConfig
     * @return
     */
    public Properties createProducerConfig(){
        Properties properties = new Properties();
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("bootstrap.servers", "bigdata-pro-m04:9092");
        return properties;
    }
    @Override
    public void run(){
        int counter = 0;
        while (true){
            // 生成1000条访问数据
            for (int i = 0;i < 1000;i++){
                String log = null;
                // 生成条访问数据
                if (newOldUserArr[random.nextInt(10)] == 1){
                    log = getAccessLog();
                } else {
                    log = getRegisterLog();
                }
                // 将数据发送给kafka
                producer.send(new ProducerRecord<String, String>(topic,log));
                counter++;
                if (counter == 100){
                    counter = 0;
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
    /**
     * 生成注册数据
     * @return
     */
    private static String getRegisterLog(){
        StringBuffer stringBuffer = new StringBuffer("");
        // 生成时间戳
        long timestamp = System.currentTimeMillis();
        // 随机生成userid（默认1000注册用户，每天1/10的访客是未注册用户）
        Long userid = 0L;
        int newOldUser = newOldUserArr[random.nextInt(10)];
        if (newOldUser == 1){
            userid = null;
        } else {
            userid = (long)random.nextInt(100000);
        }
        // 随机生成pageid，共1000个页面
        long pageid = random.nextInt(1000);
        // 随机生成板块
        String section = sections[random.nextInt(sections.length)];
        // 生成固定的行为，view
        String action = "view";
        return stringBuffer.append(date).append(" ")
                .append(timestamp).append(" ")
                .append(userid).append(" ")
                .append(pageid).append(" ")
                .append(section).append(" ")
                .append(action).toString();
    }
    /**
     * 生成访问数据
     * @return
     */
    private static String getAccessLog(){
        StringBuffer stringBuffer = new StringBuffer("");
        // 生成时间戳
        long timestamp = System.currentTimeMillis();
        // 新用户都是userid为null
        Long userid = null;
        // 生成随机pageid，都是null
        Long pageid = null;
        // 生成随机版块，都是null
        String section = null;
        // 生成固定的行为，view
        String action = "register";
        return stringBuffer.append(date).append(" ")
                .append(timestamp).append(" ")
                .append(userid).append(" ")
                .append(pageid).append(" ")
                .append(section).append(" ")
                .append(action).toString();
    }
    public static void main(String[] args) {
        AccessProducer accessProducer = new AccessProducer("spark");
        accessProducer.start();
    }
}

package com.kfk.spark.common;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
/**
 * @author : 蔡政洁
 * @email :caizhengjie888@icloud.com
 * @date : 2020/12/14
 * @time : 8:23 下午
 */
public class CommStreamingContext {
    public static JavaStreamingContext getJssc(){
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("CommStreamingContext");
        return new JavaStreamingContext(conf, Durations.seconds(5));
    }
}

package com.kfk.spark.news_analysis_project;
import com.kfk.spark.common.CommStreamingContext;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.spark.api.java.JavaRDD;
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.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import scala.Tuple2;
import java.util.*;
/**
 * @author : 蔡政洁
 * @email :caizhengjie888@icloud.com
 * @date : 2020/12/20
 * @time : 4:11 下午
 */
public class NewsRealTime {
    /**
     * input data:
     * 2020-12-20 1608451521565 364 422 fashion view
     * 2020-12-20 1608451521565 38682 708 aviation view
     * ...
     * @param args
     */
    public static void main(String[] args) throws InterruptedException {
        JavaStreamingContext jssc = CommStreamingContext.getJssc();
        // sparkstreaming与kafka连接
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", "bigdata-pro-m04:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "streaming_kafka_1");
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", false);
        // 设置topic
        Collection<String> topics = Collections.singletonList("spark");
        // kafka数据源
        JavaInputDStream<ConsumerRecord<String, String>> stream =
                KafkaUtils.createDirectStream(
                        jssc,
                        LocationStrategies.PreferConsistent(),
                        ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams)
                );
        /**
         * stream -> map -> JavaDStream
         */
        JavaDStream<String> accessDstream = stream.map(new Function<ConsumerRecord<String, String>, String>() {
            @Override
            public String call(ConsumerRecord<String, String> v1) throws Exception {
                return v1.value();
            }
        });
        /**
         * accessDStream -> filter -> action(view)
         */
        JavaDStream<String> filterDstream = accessDstream.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String v1) throws Exception {
                String[] lines = v1.split(" ");
                String action = lines[5];
                String actionValue = "view";
                if (actionValue.equals(action)){
                    return true;
                } else {
                    return false;
                }
            }
        });
        // 求网页的pv
        calculatePagePV(filterDstream);
        // 求网页的uv
        calculatePageUV(filterDstream);
        // 求注册用户数
        calculateRegistercount(accessDstream);
        // 求热门板块
        calculateUserSectionPV(accessDstream);
        jssc.start();
        jssc.awaitTermination();
    }
    /**
     * 求网页的pv
     * input data:
     * 2020-12-20 1608451521565 364 422 fashion view
     *
     * 数据演化过程：
     * filterDstream -> mapToPair -> <2020-12-20_422,1> -> reduceByKey -> <2020-12-20_422,5>
     *
     * @param filterDstream
     */
    public static void calculatePagePV(JavaDStream<String> filterDstream){
        /**
         * filterDstream -> mapToPair -> <2020-12-20_422,1>
         */
        JavaPairDStream<String,Integer> pairDstream = filterDstream.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String,Integer> call(String lines) throws Exception {
                String[] line = lines.split(" ");
                return new Tuple2<String,Integer>(line[0] + "_" + line[3], 1);
            }
        });
        /**
         * pairDstream -> reduceByKey -> <2020-12-20_422,5>
         */
        JavaPairDStream<String,Integer> pvStream = pairDstream.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1+v2;
            }
        });
        pvStream.print();
        /**
         * (2020-12-21_16,1)
         * (2020-12-21_548,1)
         * (2020-12-21_881,1)
         * (2020-12-21_27,1)
         * (2020-12-21_771,1)
         * (2020-12-21_344,2)
         * (2020-12-21_313,1)
         * (2020-12-21_89,1)
         * (2020-12-21_14,1)
         * (2020-12-21_366,1)
         * ...
         */
    }
    /**
     * 求网页的uv
     * input data:
     * 2020-12-20 1608451521565 364 422 fashion view
     * 2020-12-20 1608451521565 364 422 fashion view
     * 2020-12-20 1608451521565 365 422 fashion view
     * 2020-12-20 1608451521565 366 422 fashion view
     * 2020-12-20 1608451521565 367 422 fashion view
     * 2020-12-20 1608451521565 367 453 fashion view
     *
     * 数据演化过程：
     * 第一步：map
     * (2020-12-20,364,422)
     * (2020-12-20,364,422)
     * (2020-12-20,365,422)
     * (2020-12-20,366,422)
     * (2020-12-20,367,422)
     * (2020-12-20,367,453)
     *
     * 第二步：rdd -> distinct
     * (2020-12-20,364,422)
     * (2020-12-20,365,422)
     * (2020-12-20,366,422)
     * (2020-12-20,367,422)
     * (2020-12-20,367,453)
     *
     * 第三步：mapToPair
     * <2020-12-20_422,1>
     * <2020-12-20_422,1>
     * <2020-12-20_422,1>
     * <2020-12-20_422,1>
     * <2020-12-20_453,1>
     *
     * 第四步：reduceByKey
     * <2020-12-20_422,4>
     * <2020-12-20_453,1>
     *
     * @param filterDstream
     */
    public static void calculatePageUV(JavaDStream<String> filterDstream){
        /**
         * filterDstream -> map -> (2020-12-20,364,422)
         */
        JavaDStream<String> mapDstream = filterDstream.map(new Function<String, String>() {
            @Override
            public String call(String lines) throws Exception {
                String[] line = lines.split(" ");
                return line[0] + "," + line[2] + "," + line[3];
            }
        });
        /**
         * mapDstream -> distinct
         */
        JavaDStream<String> distinctDstream = mapDstream.transform(new Function<JavaRDD<String>, JavaRDD<String>>() {
            @Override
            public JavaRDD<String> call(JavaRDD<String> lines) throws Exception {
                return lines.distinct();
            }
        });
        /**
         * distinctDstream -> mapToPair -> <2020-12-20_422,1>
         */
        JavaPairDStream<String,Integer> pairDstream = distinctDstream.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String lines) throws Exception {
                String[] line = lines.split(",");
                return new Tuple2<>(line[0] + "_" + line[2], 1);
            }
        });
        /**
         * pairDstream -> reduceByKey -> <2020-12-20_422,4>
         */
        JavaPairDStream<String,Integer> uvStream = pairDstream.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1+v2;
            }
        });
        uvStream.print();
        /**
         * (2020-12-21_492,1)
         * (2020-12-21_85,2)
         * (2020-12-21_18,1)
         * (2020-12-21_27,2)
         * (2020-12-21_825,1)
         * (2020-12-21_366,1)
         * (2020-12-21_89,1)
         * (2020-12-21_14,2)
         * (2020-12-21_69,1)
         * (2020-12-21_188,1)
         * ...
         */
    }
    /**
     * 求注册用户数:过滤出action=register的数据就可以
     * input data:
     * 2020-12-20 1608451521565 364 422 fashion view
     *
     * 数据演化过程：
     * accessDStream -> filter -> action(register) -> mapToPair -> reduceByKey
     *
     * @param accessDstream
     */
    public static void calculateRegistercount(JavaDStream<String> accessDstream){
        /**
         * accessDStream -> filter -> action(register)
         */
        JavaDStream<String> filterDstream = accessDstream.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String v1) throws Exception {
                String[] lines = v1.split(" ");
                String action = lines[5];
                String actionValue = "register";
                if (actionValue.equals(action)){
                    return true;
                } else {
                    return false;
                }
            }
        });
        /**
         * filterDstream -> mapToPair -> <2020-12-20_register,1>
         */
        JavaPairDStream<String,Integer> pairDstream = filterDstream.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String,Integer> call(String lines) throws Exception {
                String[] line = lines.split(" ");
                return new Tuple2<String,Integer>(line[0] + "_" + line[5], 1);
            }
        });
        /**
         * pairDstream -> reduceByKey -> <2020-12-20_register,5>
         */
        JavaPairDStream<String,Integer> registerCountStream = pairDstream.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1+v2;
            }
        });
        registerCountStream.print();
        /**
         * (2020-12-21_register,11)
         */
    }
    /**
     * 求出热门板块
     * input data:
     * 2020-12-20 1608451521565 364 422 fashion view
     *
     * 数据演化过程：
     * filterDstream -> mapToPair -> <2020-12-20_fashion,1> -> reduceByKey -> <2020-12-20_fashion,5>
     *
     * @param filterDstream
     */
    public static void calculateUserSectionPV(JavaDStream<String> filterDstream){
        /**
         * filterDstream -> mapToPair -> <2020-12-20_fashion,1>
         */
        JavaPairDStream<String,Integer> pairDstream = filterDstream.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String,Integer> call(String lines) throws Exception {
                String[] line = lines.split(" ");
                return new Tuple2<String,Integer>(line[0] + "_" + line[4], 1);
            }
        });
        /**
         * pairDstream -> reduceByKey -> <2020-12-20_fashion,5>
         */
        JavaPairDStream<String,Integer> pvStream = pairDstream.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1+v2;
            }
        });
        pvStream.print();
        /**
         * (2020-12-21_internet,16)
         * (2020-12-21_military,24)
         * (2020-12-21_aviation,21)
         * (2020-12-21_carton,19)
         * (2020-12-21_government,25)
         * (2020-12-21_tv-show,19)
         * (2020-12-21_country,14)
         * (2020-12-21_movie,13)
         * (2020-12-21_international,16)
         * ...
         */
    }
}