每日top3热点搜索词统计案例

数据格式：

日期，用户，搜索词，平台，版本

需求：

1、筛选出符合条件（城市，平台，版本）的数据

2、统计每天搜索uv排名前三的搜索词

3、按照每天的top3搜索词的uv搜索总次数，倒叙排序

4、将数据保存到hive表中

思路分析

1、针对原始数据（HDFS文件），获取输入RDD

2、使用filter算法，针对输入RDD中的数据，进行数据过滤，过滤出符合条件的数据

2．1普通的算法：直接在filter算法函数中，使用外部的查询条件（map），但是，这样做的话，是不是查询条件map，会发送到每一个task上一份副本，（性能不好）

2.2优化后的做法，将查询条件，封装为broadCast广播变量，在filter算法中使用broadCast广播变量。

3、将数据转换为（日期_搜索词，用户）格式，对他进行分组。然后再次进行映射，对每天每个搜索词的搜索用户进行去重操作，并统计去重后的数据，即为每天每个搜搜词的uv，最后获得（日期_搜搜词，uv）。

4、将得到的每天每个搜索词的uvRDD，映射为元素类型的RowRDD，转换为DataFrame。

5、注册为临时表，使用SparkSQL的开窗函数，统计每天的uv数量排名前三名的搜索词，以及他的搜索nv，最后获得一个DataFrame

6、将DataFrame转换为RDD，继续操作，按照每天日期来分组，并进行映射，计算出每天的top3搜索词的uv的总数，然后将uv总数作为key，将每天的top3搜索词以及搜索次数，拼接为一个字符串

7、按照每天的top3搜索总uv，进行排序，倒序排序

8、将排好的数据，再次映射回来，变成日期_搜索词_uv的格式

9、再次映射为DataFrame，并将数据保存到hive中。

文本：

2018-10-1:leo:water:beijing:android:1.0
2018-10-1:leo1:water:beijing:android:1.0
2018-10-1:leo2:water:beijing:android:1.0
2018-10-1:jack:water:beijing:android:1.0
2018-10-1:jack1:water:beijing:android:1.0
2018-10-1:leo:seafood:beijing:android:1.0
2018-10-1:leo1:seafood:beijing:android:1.0
2018-10-1:leo2:seafood:beijing:android:1.0
2018-10-1:leo:food:beijing:android:1.0
2018-10-1:leo1:food:beijing:android:1.0
2018-10-1:leo2:meat:beijing:android:1.0
2018-10-2:leo:water:beijing:android:1.0
2018-10-2:leo1:water:beijing:android:1.0
2018-10-2:leo2:water:beijing:android:1.0
2018-10-2:jack:water:beijing:android:1.0
2018-10-2:leo1:seafood:beijing:android:1.0
2018-10-2:leo2:seafood:beijing:android:1.0
2018-10-2:leo3:seafood:beijing:android:1.0
2018-10-2:leo1:food:beijing:android:1.0
2018-10-2:leo2:food:beijing:android:1.0
2018-10-2:leo:meat:beijing:android:1.0

代码：

package com.bynear.spark_sql;

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.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;

import java.util.*;

public class DailyTop3Keyword {public static void main(String[] args) {SparkConf conf = new SparkConf();
        JavaSparkContext jsc = new JavaSparkContext(conf);
        SQLContext sqlContext = new HiveContext(jsc.sc());
//        伪造数据(这些数据可以来自mysql数据库)
        final HashMap<String, List<String>> queryParaMap = new HashMap<String, List<String>>();
        queryParaMap.put("city", Arrays.asList("beijing"));
        queryParaMap.put("platform", Arrays.asList("android"));
        queryParaMap.put("version", Arrays.asList("1.0", "1.2", "2.0", "1.5"));
//      将数据进行广播
        final Broadcast<HashMap<String, List<String>>> queryParamMapBroadcast = jsc.broadcast(queryParaMap);
//      读取文本
        JavaRDD<String> rowRDD = jsc.textFile("hdfs://Spark01:9000/zjs/daily.txt");
//      filter算子进行过滤
        JavaRDD<String> filterRDD = rowRDD.filter(new Function<String, Boolean>() {@Override
            public Boolean call(String log) throws Exception {String[] logSplit = log.split(":");
                String city = logSplit[3];
                String platform = logSplit[4];
                String version = logSplit[5];
                HashMap<String, List<String>> queryParamMap = queryParamMapBroadcast.value();
                List<String> cities = queryParamMap.get("city");
                if (!cities.contains(city) && cities.size() > 0) {return false;
                }List<String> platforms = queryParamMap.get("platform");
                if (!platforms.contains(platform)) {return false;
                }List<String> versions = queryParamMap.get("version");
                if (!versions.contains(version)) {return false;
                }return true;
            }});
//        过滤出来的原始日志，映射为（日期_搜索词,用户）格式
        JavaPairRDD<String, String> dateKeyWordUserRDD = filterRDD.mapToPair(new PairFunction<String, String, String>() {@Override
            public Tuple2<String, String> call(String log) throws Exception {String[] logSplit = log.split(":");
                String date = logSplit[0];
                String user = logSplit[1];
                String keyword = logSplit[2];
                return new Tuple2<String, String>(date + "_" + keyword, user);
            }});
//        进行分组，获取每天每个搜索词，有哪些用户搜索了（没有去重）
        JavaPairRDD<String, Iterable<String>> dateKeywordUsersRDD = dateKeyWordUserRDD.groupByKey();
        List<Tuple2<String, Iterable<String>>> collect1 = dateKeywordUsersRDD.collect();
        for (Tuple2<String, Iterable<String>> tuple2 : collect1) {System.out.println("进行分组，获取每天每个搜索词，有哪些用户搜索了（没有去重）" + tuple2._2);
            System.out.println(tuple2);
        }//        对每天每个搜索词的搜索用户  去重操作  获得前uv
        JavaPairRDD<String, Long> dateKeywordUvRDD = dateKeywordUsersRDD.mapToPair(new PairFunction<Tuple2<String, Iterable<String>>, String, Long>() {@Override
                    public Tuple2<String, Long> call(Tuple2<String, Iterable<String>> dataKeywordUsers) throws Exception {String dateKeyword = dataKeywordUsers._1;
                        Iterator<String> users = dataKeywordUsers._2.iterator();
//                去重   并统计去重后的数量
                        List<String> distinctUsers = new ArrayList<String>();
                        while (users.hasNext()) {String user = users.next();
                            if (!distinctUsers.contains(user)) {distinctUsers.add(user);
                            }}
//              获取uv
                        long uv = distinctUsers.size();
//                日期_搜索词，用户个数
                        return new Tuple2<String, Long>(dateKeyword, uv);
                    }});
        List<Tuple2<String, Long>> collect2 = dateKeywordUvRDD.collect();
        for (Tuple2<String, Long> stringLongTuple2 : collect2) {System.out.println("对每天每个搜索词的搜索用户  去重操作  获得前uv");
            System.out.println(stringLongTuple2);
        }//        将每天每个搜索词的uv数据，转换成DataFrame
        JavaRDD<Row> dateKeywordUvRowRDD = dateKeywordUvRDD.map(new Function<Tuple2<String, Long>, Row>() {@Override
            public Row call(Tuple2<String, Long> dateKeywordUv) throws Exception {String date = dateKeywordUv._1.split("_")[0];
                String keyword = dateKeywordUv._1.split("_")[1];
                long uv = dateKeywordUv._2;
                return RowFactory.create(date, keyword, uv);
            }});
        ArrayList<StructField> fields = new ArrayList<StructField>();
        fields.add(DataTypes.createStructField("date", DataTypes.StringType, true));
        fields.add(DataTypes.createStructField("keyword", DataTypes.StringType, true));
        fields.add(DataTypes.createStructField("uv", DataTypes.LongType, true));
        StructType structType = DataTypes.createStructType(fields);
        DataFrame dateKeywordUvDF = sqlContext.createDataFrame(dateKeywordUvRowRDD, structType);
        dateKeywordUvDF.registerTempTable("sales");
//        使用开窗函数，统计每天搜索uv排名前三的热点搜索词
//        日期  搜索词   人数个数  前三名
        final DataFrame dailyTop3KeyWordDF = sqlContext.sql("select date,keyword,uv from (select date, keyword, uv, row_number() over (partition by date order by uv DESC ) rank from sales ) tmp_sales where rank <=3");
//        将DataFrame转换为RDD， 映射，
        JavaRDD<Row> dailyTop3KeyWordRDD = dailyTop3KeyWordDF.javaRDD();

        JavaPairRDD<String, String> dailyTop3KeywordRDD = dailyTop3KeyWordRDD.mapToPair(new PairFunction<Row, String, String>() {@Override
            public Tuple2<String, String> call(Row row) throws Exception {String date = String.valueOf(row.get(0));
                String keyword = String.valueOf(row.get(1));
                String uv = String.valueOf(row.get(2));
//                映射为  日期  搜索词_总个数
                return new Tuple2<String, String>(date, keyword + "_" + uv);
            }});

        List<Tuple2<String, String>> collect = dailyTop3KeywordRDD.collect();
        for (Tuple2<String, String> stringStringTuple2 : collect) {System.out.println("开窗函数操作");
            System.out.println(stringStringTuple2);
        }//          根据 日期分组
        JavaPairRDD<String, Iterable<String>> top3DateKeywordsRDD = dailyTop3KeywordRDD.groupByKey();
//        进行映射
        JavaPairRDD<Long, String> uvDateKeywordsRDD = top3DateKeywordsRDD.mapToPair(new PairFunction<Tuple2<String, Iterable<String>>, Long, String>() {@Override
            public Tuple2<Long, String> call(Tuple2<String, Iterable<String>> tuple) throws Exception {String date = tuple._1;
//                搜索词_总个数  集合
                Iterator<String> KeyWordUviterator = tuple._2.iterator();
                long totalUv = 0L;
                String dateKeyword = date;
                while (KeyWordUviterator.hasNext()) {
//                    搜索词_个数
                    String keywoarUv = KeyWordUviterator.next();
                    Long uv = Long.valueOf(keywoarUv.split("_")[1]);
                    totalUv += uv;
                    dateKeyword = dateKeyword + "," + keywoarUv;
                }return new Tuple2<Long, String>(totalUv, dateKeyword);
            }});
        JavaPairRDD<Long, String> sortedUvDateKeywordsRDD = uvDateKeywordsRDD.sortByKey(false);
        List<Tuple2<Long, String>> rows = sortedUvDateKeywordsRDD.collect();
        for (Tuple2<Long, String> row : rows) {System.out.println(row._2 + "    " + row._1);
        }//        映射
        JavaRDD<Row> resultRDD = sortedUvDateKeywordsRDD.flatMap(new FlatMapFunction<Tuple2<Long, String>, Row>() {@Override
            public Iterable<Row> call(Tuple2<Long, String> tuple) throws Exception {String dateKeywords = tuple._2;
                String[] dateKeywordsSplit = dateKeywords.split(",");
                String date = dateKeywordsSplit[0];
                ArrayList<Row> rows = new ArrayList<Row>();
                rows.add(RowFactory.create(date, dateKeywordsSplit[1].split("_")[0],
                        Long.valueOf(dateKeywordsSplit[1].split("_")[1])));

                rows.add(RowFactory.create(date, dateKeywordsSplit[2].split("_")[0],
                        Long.valueOf(dateKeywordsSplit[2].split("_")[1])));

                rows.add(RowFactory.create(date, dateKeywordsSplit[3].split("_")[0],
                        Long.valueOf(dateKeywordsSplit[3].split("_")[1])));

                return rows;
            }});
        DataFrame finalDF = sqlContext.createDataFrame(resultRDD, structType);
        List<Row> rows1 = finalDF.javaRDD().collect();
        for (Row row : rows1) {System.out.println(row);
        }jsc.stop();

    }
}

注意点：

1、如果文本案例使用的是txt编辑，将文本保存ANSI格式，否则在groupByKey的时候，第一行默认会出现一个空格，分组失败。最开始使用的是UTF-8格式

2、文本的最后禁止出现空行，否则在split的时候会报错，出现数组越界的错误。

3、使用到窗口函数的时候，必须使用到HiveContext方法，HiveContext使用到的是SparkContext，使用使用jsc.sc()

运行结果：

进行分组，获取每天每个搜索词，有哪些用户搜索了（没有去重）[leo]
(2018-10-2_meat,[leo])
进行分组，获取每天每个搜索词，有哪些用户搜索了（没有去重）[leo, leo1, leo2, jack]
(2018-10-2_water,[leo, leo1, leo2, jack])
进行分组，获取每天每个搜索词，有哪些用户搜索了（没有去重）[leo, leo1]
(2018-10-1_food,[leo, leo1])
进行分组，获取每天每个搜索词，有哪些用户搜索了（没有去重）[leo1, leo2, leo3]
(2018-10-2_seafood,[leo1, leo2, leo3])
进行分组，获取每天每个搜索词，有哪些用户搜索了（没有去重）[leo, leo1, leo2, jack, jack1]
(2018-10-1_water,[leo, leo1, leo2, jack, jack1])
进行分组，获取每天每个搜索词，有哪些用户搜索了（没有去重）[leo2]
(2018-10-1_meat,[leo2])
进行分组，获取每天每个搜索词，有哪些用户搜索了（没有去重）[leo, leo1, leo2]
(2018-10-1_seafood,[leo, leo1, leo2])
进行分组，获取每天每个搜索词，有哪些用户搜索了（没有去重）[leo1, leo2]

(2018-10-2_food,[leo1, leo2])

对每天每个搜索词的搜索用户去重操作获得前uv
(2018-10-2_meat,1)
对每天每个搜索词的搜索用户去重操作获得前uv
(2018-10-2_water,4)
对每天每个搜索词的搜索用户去重操作获得前uv
(2018-10-1_food,2)
对每天每个搜索词的搜索用户去重操作获得前uv
(2018-10-2_seafood,3)
对每天每个搜索词的搜索用户去重操作获得前uv
(2018-10-1_water,5)
对每天每个搜索词的搜索用户去重操作获得前uv
(2018-10-1_meat,1)
对每天每个搜索词的搜索用户去重操作获得前uv
(2018-10-1_seafood,3)
对每天每个搜索词的搜索用户去重操作获得前uv

(2018-10-2_food,2)

窗函数操作
(2018-10-1,water_5)
开窗函数操作
(2018-10-1,seafood_3)
开窗函数操作
(2018-10-1,food_2)
开窗函数操作
(2018-10-2,water_4)
开窗函数操作
(2018-10-2,seafood_3)
开窗函数操作

(2018-10-2,food_2)

最终结果

[2018-10-1,water,5]
[2018-10-1,seafood,3]

[2018-10-1,food,2]

[2018-10-2,water,4]
[2018-10-2,seafood,3]
[2018-10-2,food,2]