学习目标/Target

掌握各区域热门商品Top3分析实现思路

掌握如何创建Spark连接并读取数据集

掌握利用Spark获取业务数据

掌握利用Spark过滤商品的行为类型

掌握利用Spark转换数据格式

掌握利用Spark统计每个区域中的不同商品

掌握利用Spark根据区域进行分组

掌握利用Spark根据区域内商品的查看次数进行排序

掌握将数据持久化到HBase数据库

熟悉通过Spark On YARN运行程序

       用户在访问电商网站时，网站在存储用户行为数据的同时，还会通过IP地址或位置信息存储用户触发行为所在的区域数据。通过统计各区域不同商品被查看的次数，获取每个区域内比较热门的商品。本章将通过对电商网站存储的用户行为数据进行分析，从而统计出各区域排名前3的热门商品。

1. 实现思路分析

       获取数据集中所有用户数据，过滤出用户行为类型为查看的数据，通过商品被查看的次数为依据判断哪些商品属于热门商品。对过滤后的数据进行聚合操作，统计每个区域不同商品的查看次数。按照区域对聚合后的数据进行分组处理，将分组后的数据进行降序排序，获取各区域排名前3的商品，就是各区域热门商品Top3。

读取/转换：读取数据集中的区域名称（address_name）、行为类型（event_type）和商品ID（product_id）数据；

过滤：过滤行为类型为view（查看）的数据；

转换：便于后续聚合处理时，将相同Key的Value值进行累加，这里需要对数据格式进行转换处理，将区域名称和商品ID作为Key，值1作为Value。由于过滤后的数据行为类型都是查看，在后续的处理中便不再需要行为类型数据；

聚合：统计每个区域中不同商品的查看次数；

转换/分组：对数据格式进行转换，将区域名称作为Key，商品ID和商品被查看的次数作为Value。接下来，将转换后的数据根据Key进行分组，统计各个区域被查看的商品及每个商品查看的次数；

排序：对每一组数据的值进行排序，即对各个区域每个商品被查看的次数进行降序排序。

2. 实现各区域热门商品Top3

2.1  创建Spark连接并读取数据集

       在项目SparkProject的 java目录新建Package包“cn.itcast.top3”，用于存放实现各区域热门商品Top3的Java文件。在包“cn.itcast.top3”中创建文件AreaProductTop3.java，用于实现各区域热门商品Top3。

public class AreaProductTop3{
    public static void main(String[] arg){
    //实现各区域热门商品Top3分析
    }f
}

       在main()方法中，创建JavaSparkContext和SparkConf对象，JavaSparkContext对象用于实现Spark程序，SparkConf对象用于配置Spark程序相关参数。        

SparkConf conf = new SparkConf(); //设置Application名称为top3_area_product conf.setAppName("top3_area_product"); JavaSparkContext sc = new JavaSparkContext(conf);

       在main()方法中，调用JavaSparkContext对象的textFile()方法读取外部文件，将文件中的数据加载到textFileRDD。

JavaRDD<String> textFileRDD = sc.textFile(arg[0]);

2.2 获取业务数据

       在main()方法中，使用mapToPair()算子转换textFileRDD的每一行数据，用于获取每一行数据中的行为类型、区域名称和商品ID数据，将转换结果加载到transProductRDD。

JavaPairRDD<Tuple2<String,String>,String> transProductRDD     = textFileRDD.mapToPair(new PairFunction<String,Tuple2<String, String>,String>() {    

@Override    

public Tuple2<Tuple2<String, String>, String> call(String s) throws Exception {

       JSONObject json = JSONObject.parseObject(s);

       String address_name = json.getString("address_name").replaceAll("\\u00A0+","");                 String product_id = json.getString("product_id");

       String event_type = json.getString("event_type");

       Tuple2<Tuple2<String,String>,String> tuple2 =                

       new Tuple2<>( new Tuple2<>(address_name,product_id), event_type);

       return tuple2;        

       }    

 });

2.3 过滤商品的行为类型

       在main()方法中，使用filter()算子过滤transProductRDD每一行数据中行为类型为加入购物车和购买的数据，只保留行为类型为查看的数据，将过滤结果加载到getViewRDD。

JavaPairRDD<Tuple2<String, String>, String> getViewRDD =

   transProductRDD.filter(new Function<Tuple2<

                           Tuple2<String, String>, String>, Boolean>() {

@Override

       public Boolean call(

               Tuple2<Tuple2<String, String>, String> tuple2)

               throws Exception {

           String event_type = tuple2._2;

           return event_type.equals("view");

        }    

});

2.4  转换数据格式

       在main()方法中，使用mapToPair()算子转换getViewRDD的每一行数据，用于替换行为类型数据为1，将转换结果加载到productByAreaRDD。

JavaPairRDD<Tuple2<String,String>,Integer> productByAreaRDD =            getViewRDD.mapToPair(

              new PairFunction<Tuple2<Tuple2<String, String>, String>,

                        Tuple2<String, String>,

                        Integer>() {

@Override

      public Tuple2<Tuple2<String, String>, Integer> call(

              Tuple2<Tuple2<String, String>, String> tuple2)

              throws Exception {

          return new Tuple2<>(tuple2._1,new Integer(1));

      }

  });

2.5  统计每个区域中的不同商品

       在main()方法中，使用reduceByKey()算子对productByAreaRDD进行聚合操作，用于统计每个区域中不同商品的查看次数，将统计结果加载到productCountByAreaRDD。

JavaPairRDD<Tuple2<String,String>,Integer> productCountByAreaRDD =         productByAreaRDD.reduceByKey(

               new Function2<Integer, Integer, Integer>() {

@Override

   public Integer call(Integer integer, Integer integer2)

           throws Exception {

       return integer+integer2;

   }

});

2.6 根据区域进行分组

       在main()方法中，使用mapToPair()算子转换productCountByAreaRDD的每一行数据，将转换结果加载到transProductCountByAreaRDD

JavaPairRDD<String,Tuple2<String,Integer>> transProductCountByAreaRDD =productCountByAreaRDD.mapToPair(new PairFunction<Tuple2<Tuple2<String, String>, Integer>,String, Tuple2<String, Integer>>() {            

@Override

public Tuple2<String, Tuple2<String, Integer>> call(Tuple2<Tuple2<String, String>, Integer> tuple2) throws Exception {
                return new Tuple2<>(tuple2._1._1, new Tuple2<>(tuple2._1._2,tuple2._2));
            }
        });

       在main()方法中，使用groupByKey()算子对transProductCountByAreaRDD进行分组操作，将同一区域内的商品以及商品被查看的次数合并在一起，通过productGroupByAreaRDD加载分组结果。

JavaPairRDD<String, Iterable<Tuple2<String, Integer>>>

        productGroupByAreaRDD = transProductCountByAreaRDD.groupByKey();

2.7 根据区域内商品的查看次数进行排序

       在main()方法中，使用mapToPair()算子转换productGroupByAreaRDD的每一行数据，将同一区域内的商品按照商品被查看的次数进行逆序排序，通过productSortByAreaRDD加载排序结果。

JavaPairRDD<String, Iterable<Tuple2<String, Integer>>> productSortByAreaRDD =productGroupByAreaRDD

          .mapToPair( new PairFunction<Tuple2<String, Iterable<Tuple2<String, Integer>>>,String,Iterable<Tuple2<String, Integer>>>() {

@Override

   public Tuple2<String, Iterable<Tuple2<String, Integer>>> call(Tuple2<String, Iterable<Tuple2<String, Integer>>> tuple2)

            throws Exception {

                                         List<Tuple2<String,Integer>> list = new ArrayList<>();                                                              Iterator<Tuple2<String,Integer>> iter = tuple2._2.iterator();                                                              while (iter.hasNext()){

                                        list.add(iter.next());

                  }

                  list.sort(new Comparator<Tuple2<String, Integer>>() {

@Override

                  public int compare(Tuple2<String, Integer> o1,Tuple2<String, Integer> o2) {

       return o2._2 - o1._2;

}    

});  

return new Tuple2<>(tuple2._1,list);

   }          

});

2.8 数据持久化

获取各区域热门商品Top3数据

在类AreaProductTop3的main()方法中，使用mapToPair()算子转换productSortByAreaRDD的每一行数据，获取每个区域排名前3的商品，通过productSortByAreaRDD加载转换结果。

JavaPairRDD<String, Iterable<Tuple2<String, Integer>>> top3AreaProductRDD=productSortByAreaRDD.mapToPair(new PairFunction<Tuple2<String, Iterable<Tuple2<String, Integer>>>,String,Iterable<Tuple2<String, Integer>>>() {    

@Override

   public Tuple2<String, Iterable<Tuple2<String, Integer>>> call(Tuple2<String, Iterable<Tuple2<String, Integer>>> tuple2) throws Exception {

       List<Tuple2<String,Integer>> list = new ArrayList<>();

       Iterator<Tuple2<String,Integer>> iter = tuple2._2.iterator();

       int i = 0;

       while (iter.hasNext()){

           list.add(iter.next());

           i++;

           if (i == 3){

               break;

           }

       }

       return new Tuple2<>(tuple2._1,list);

   }

});

持久化各区域热门商品Top3数据

       在类AreaProductTop3的main()方法中，添加方法top3ToHbase()，用于将各区域热门商品Top3分析结果持久化到HBase数据库中，该方法包含参数rdd，表示各区域热门商品Top3分析结果数据。

public static void top3ToHbase(JavaPairRDD<String, Iterable<Tuple2<String, Integer>>> rdd) throws IOException {

}

       在方法top3ToHbase()中创建数据表top3和列族top3_area_product，并且创建数组column用于存储数据表top3的列名。

HbaseUtils.createTable("top3","top3_area_product");  

String[] column ={"area","product_id","viewcount"};

持久化各区域热门商品Top3数据

在方法top3ToHbase()中通过foreach()算子遍历各区域热门商品Top3分析结果数据。

rdd.foreach(new VoidFunction<Tuple2<String,Iterable<Tuple2<String, Integer>>>>()

       {

@Override

public void call(Tuple2<String, Iterable<Tuple2<String, Integer>>> tuple2) throws Exception {                 String area = tuple2._1,product_id = "",viewcount = "";

               Iterator<Tuple2<String,Integer>> iter = tuple2._2.iterator();

               List<Tuple2<String,Integer>> myList = Lists.newArrayList(iter);

               for (Tuple2<String,Integer> tuple : myList) {

                   product_id = tuple._1;

                   viewcount = String.valueOf(tuple._2);

                   String [] value =  {area,product_id,viewcount};

                   try {                                 HbaseUtils.putsToHBase("top3",area+product_id,"top3_area_product",column,value);                     } catch (Exception e) {

                       e.printStackTrace();

                   }

               }

           }

       });

}

       在类AreaProductTop3的main()方法中，调用top3ToHbase()方法并传入参数top3AreaProductRDD，用于在Spark程序中实现top3ToHbase()方法，将各区域热门商品Top3分析结果持久化到HBase数据库中的数据表top3。

try {

   top3ToHbase(top3AreaProductRDD);

       } catch (IOException e) {

   e.printStackTrace();

}

HbaseConnect.closeConnection();

sc.close();

3. 运行程序

       在IntelliJ IDEA中将各区域热门商品Top3分析程序封装成jar包，并上传到集群环境中，通过spark-submit将程序提交到YARN中运行。

封装jar包：

       由于在封装热门品类Top10分析程序jar包时，将程序主类指向了“cn.itcast.top10.CategoryTop10”，因此这里需要将pom.xml文件中的程序主类修改为“cn.itcast.top3.AreaProductTop3”。根据封装热门品类Top10分析程序jar包的方式封装各区域热门商品Top3分析程序。将封装完成的jar包重命名为“AreaProductTop3”，通过远程连接工具SecureCRT将AreaProductTop3.jar上传到虚拟机Spark01的/export/SparkJar/目录下。

提交各区域热门商品Top3分析程序到YARN集群

通过Spark安装目录中bin目录下的shell脚本文件spark-submit提交各区域热门商品Top3分析程序到YARN集群运行。

查看程序运行结果：

在虚拟机Spark01执行“hbase shell”命令，进入HBase命令行工具。

在HBase命令行工具中执行“list”命令，查看HBase数据库中的所有数据表。

> list TABLE                                                                                      

test                                                                                        

top10 top3                                                                                

2 row(s) in 0.1810 seconds

在HBase命令行工具执行“scan 'top3'”命令，查看数据表top3中所有数据。

       本文主要讲解了如何通过用户行为数据实现各区域热门商品Top3分析，首先通过分析实现思路，使读者了解各区域热门商品Top3分析的实现流程。然后通过IntelliJ IDEA开发工具实现各区域热门商品Top3分析程序并将分析结果存储到HBase数据库，使读者掌握运用Java语言编写Spark Core和HBase程序的能力。最后封装各区域热门商品Top3分析程序并提交到集群运行，使读者掌握运用IntelliJ IDEA开发工具封装Spark Core程序以及Spark ON YARN模式运行Spark Core程序的方法。