Spark运用-恢复实时数据

每天学Java

发布于 2020-06-01 10:43:51

4890

发布于 2020-06-01 10:43:51

业务背景

目前在个人的网站和小程序中上线了文章模块，于是想在网站的功能集锦模块中将每日文章访问的实时数据进行展示，由于当作一个小的功能集成到网站中，所以并没有使用太多的中间件，只是单纯的在拦截器那里加一些逻辑将文章访问数据存入到内存中(并不是Redis)，然后提供接口将数据在页面上进行展示。于是就出现了一个问题，由于是小功能嵌入到网站上，更改重启很频繁，导致每次重启今日文章数据都会丢失，最终想到利用Spark来恢复每日数据。

呈现效果

网址:www.study-java.cn

微信小程序:每天学Java

解决过程

1.在存入内存前，将访问数据存入日志中记录下来。

2.启动项目过程中，利用Spark解析日志，恢复日志数据。

实现过程

Order配合ApplicationRunner代表启动执行run方法。

@Component
@Order(3)
public class AnalysedArticle implements ApplicationRunner, Serializable {
    private Logger logger = LoggerFactory.getLogger(AnalysedArticle.class);

    @Override
    public void run(ApplicationArguments args) throws Exception {
        logger.info("恢复数据");
        reSume();
    }
    
    //重启恢复今日的文章访问量
    public void reSume() {
        LocalDate date = LocalDate.now();
        String logFile = "file:///home/project/email/viewJar/log/" + date + ".json";
        SparkConf sparkConf = new SparkConf().setAppName("Resume").setMaster("local[4]");
        JavaSparkContext sc = null;
        try {
        sc = new JavaSparkContext(sparkConf);
        //导入文件
        JavaRDD<String> logData = sc.textFile(logFile).cache();
        
        logData.foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                logger.info(s);
                JSONObject jsonObject = JSON.parseObject(s);
                String url;
                if ((url = jsonObject.getString("url")).contains(".md")) {
                    //DataModal使用Map存储数据
                    DataModal.put(url);
                }
            }
        });
        }finally {
            sc.close();
        }

    }

    public static void main(String[] args) {
        new AnalysedArticle().reSume();
    }


}

DataModal

public class DataModal {
    //日期->文章后缀:数量
    private static Map<String, Map<String, AtomicInteger>> map;

    static {
        map = new HashMap<>();
    }


    public static void put(String url) {
        synchronized (DataModal.map) {
            LocalDate date = LocalDate.now();
            Map<String, AtomicInteger> map2 = DataModal.map.get(date.toString());
            if (map2 == null) {
                Map<String, AtomicInteger> map3 = new HashMap<>();
                DataModal.map.put(date.toString(), map3);
                DataModal.map.get(date.toString()).put(url.substring(url.lastIndexOf("/") + 1), new AtomicInteger(1));
            } else {
                AtomicInteger atomicInteger = DataModal.map.get(date.toString()).get(url.substring(url.lastIndexOf("/") + 1));
                if (atomicInteger != null) {
                    atomicInteger.incrementAndGet();
                } else {
                    DataModal.map.get(date.toString()).put(url.substring(url.lastIndexOf("/") + 1), new AtomicInteger(1));
                }

            }
        }
    }

    public static  Map<String, AtomicInteger> get(LocalDate localDate){
       return DataModal.map.get(localDate.toString());
    }
    
    public static void remove(){
       //凌晨清空数据
       map = null;
    }
}

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2019-12-24，如有侵权请联系 cloudcommunity@tencent.com 删除

腾讯云开发者社区