4.2 需求1实现

作者: yayooo | 来源:发表于2019-08-02 23:37 被阅读0次

4.2 需求1实现
需求实现1
产品经理，你应该知道什么是需求分析
Swift基于GCD实现的倒计时功能
第 4 章需求工程
理论知识
（二）父组件传递事件给子组件（通过localStorage方式）
需求决定的价格~4.2
python 实现迭代协议
centos7.2安装node.js环境

一、实现

package com.atguigu.sparkmall.offline

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util.UUID

import com.atguigu.sparkmall.common.model.CategoryTop10
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import com.atguigu.sparkmall.common.util.StringUtil


object Req1CategoryTop10Application {
  def main(args: Array[String]): Unit = {
    //需求一：获取点击、下单和支付数量排名前10的品类

    val conf = new SparkConf().setMaster("local[*]").setAppName("Req1CategoryTop10Application")
    val sc = new SparkContext(conf)

    //从文件中获取原始数据
    val lineRDD: RDD[String] = sc.textFile("input/user_visit_action.csv")

    //将数据转换结构(category_click,1),(category_order,1),(category_pay,1)
    /** category有点击品类、下单品类、付款品类*/
    val mapRDD: RDD[Array[(String, Long)]] = lineRDD.map(line => {
      val datas: Array[String] = line.split(",")
      if (datas(6) != "-1") { //-1表示无效数据
        //点击数据
        Array((datas(6) + "_click", 1L))
      } else if (StringUtil.isNotEmpty(datas(8))) {
        //下单数据
        val categoryIds: Array[String] = datas(8).split("-")
        categoryIds.map {
          id => (id + "_order", 1L)
        }
      } else if (StringUtil.isNotEmpty(datas(10))) {
        val categoryIds: Array[String] = datas(10).split("-")
        categoryIds.map {
          id => (id + "_pay", 1L)
        }
      } else {
        Array(("", 0L)) //此处是为了在前面所有条件都不满足的情况下返回和前面返回数据类型相同的结构
      }
    })

    //因为拿到的是一个数组整体mapRDD: RDD[Array[(String, Long)]]，所以需要扁平化打散成单个的tuple
    val flatmapRDD: RDD[(String, Long)] = mapRDD.flatMap(array => array)

    //因为前面当所有条件都不满足的情况下会产生("", 0L)的数据，需要过滤
    val filterRDD: RDD[(String, Long)] = flatmapRDD.filter {
      case (key, v) => {
        StringUtil.isNotEmpty(key)
      }
    }

    //将数据分组聚合(category_click,sum),(category_order,sum),(category_pay,sum)
    val reduceRDD: RDD[(String, Long)] = filterRDD.reduceByKey(_+_)

    //将数据转换结构(category,(click,sum)),(category,(order,sum)),(category,(pay,sum)),
    val mappRDD1: RDD[(String, (String, Long))] = reduceRDD.map {
      case (key, sum) => {
        val keys: Array[String] = key.split("_")
        (keys(0), (keys(1), sum))
      }
    }

    //分组，排序: 根据品类分组
    val groupRDD: RDD[(String, Iterable[(String, Long)])] = mappRDD1.groupByKey()

    //因为操作数据不方便，将数据转换程样例类RDD[(String, Iterable[(String, Long)])] --> UserVisitAction
    val taskId: String = UUID.randomUUID().toString

    val classRDD: RDD[CategoryTop10] = groupRDD.map {
      case (categoryId, iter) => {
        val map: Map[String, Long] = iter.toMap //转换成map
        CategoryTop10(taskId, categoryId, map.getOrElse("click", 0L), map.getOrElse("order", 0L), map.getOrElse("pay", 0L))
      }
    }

    //取top10
    val collectArray: Array[CategoryTop10] = classRDD.collect() //收集为Array用集合的方式进行排序

    //sortWith方便自定义排序规则
    val top10Array: Array[CategoryTop10] = collectArray.sortWith {
      (left, right) => {
        if (left.clickCount > right.clickCount) {
          true
        } else if (left.clickCount == right.clickCount) {
          if (left.orderCount > right.orderCount) {
            true
          } else if (left.orderCount == right.orderCount) {
            if (left.payCount > right.payCount) {
              true
            } else {
              false
            }
          } else {
            false
          }
        } else {
          false
        }
      }
    }.take(10)

    //将结果保存到Mysql中
    val driver = "com.mysql.jdbc.Driver"
    val url = "jdbc:mysql://linux1:3306/sparkmall-190311"
    val userName = "root"
    val passWd = "000000"

    Class.forName(driver)
    val connection: Connection = DriverManager.getConnection(url, userName, passWd)
    val sql = "insert into category_top10 ( taskId, category_id, click_count, order_count, pay_count ) values (?, ?, ?, ?, ?)"
    val statement: PreparedStatement = connection.prepareStatement(sql)

    top10Array.foreach{
      obj=>{
        statement.setObject(1, obj.taskId)
        statement.setObject(2, obj.categoryId)
        statement.setObject(3, obj.clickCount)
        statement.setObject(4, obj.orderCount)
        statement.setObject(5, obj.payCount)
        statement.executeUpdate()
      }
    }
    statement.close()
    connection.close()

    // 释放资源
    sc.stop()
  }
}

二、使用累加器进行优化

package com.atguigu.sparkmall.offline

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util.UUID

import com.atguigu.sparkmall.common.model.CategoryTop10
import com.atguigu.sparkmall.common.util.StringUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.{immutable, mutable}
import scala.collection.parallel.immutable


object Req1CategoryTop10Application1 {
  def main(args: Array[String]): Unit = {
    //需求一：获取点击、下单和支付数量排名前10的品类

    val conf = new SparkConf().setMaster("local[*]").setAppName("Req1CategoryTop10Application")
    val sc = new SparkContext(conf)

    //从文件中获取原始数据
    val lineRDD: RDD[String] = sc.textFile("input/user_visit_action.csv")

    lineRDD.foreach(println)
    println("********")

    //创建累加器
    val accumulator = new CategoryAccumulator

    //注册累加器
    sc.register(accumulator,"accumulator")

    //使用累加器
    lineRDD.foreach{
      line => {
        val datas: Array[String] = line.split(",")
        if(datas(6) != -1) {
          //点击数据
          accumulator.add(datas(6) + "_click")
        } else if (StringUtil.isNotEmpty(datas(8))) {
          //下单数据
          val categoryIds: Array[String] = datas(8).split("-")
          categoryIds.map{
            id => {
              accumulator.add(id + "_order")
            }
          }
        } else if(StringUtil.isNotEmpty(datas(10))) {
          //支付数据
          val categoryIds: Array[String] = datas(10).split("-")
          categoryIds.map{
            id => {
              accumulator.add(id + "_pay")
            }
          }
        }
      }
    }

    //获取累加器的值(category_click,sum)
    val accumulatorVal: mutable.HashMap[String, Long] = accumulator.value

    //将累加器的值根据品类进行分组 (category,(category_click,sum))
    val categoryToMap: Map[String, mutable.HashMap[String, Long]] = accumulatorVal.groupBy {
      case (k, sum) => {
        k.split("_")(0)
      }
    }

    val taskId: String = UUID.randomUUID().toString

    //将分组后的数据转换为样例类：
    val categoryTop10: Iterable[CategoryTop10] = categoryToMap.map {
      case (category, map) => {
        CategoryTop10(
          taskId,
          category,
          map.getOrElse(category + "_click", 0L),
          map.getOrElse(category + "_order", 0L),
          map.getOrElse(category + "_pay", 0L)
        )
      }
    }

    val result: List[CategoryTop10] = categoryTop10.toList.sortWith {
      (left, right) => {
        if (left.clickCount > right.clickCount) {
          true
        } else if (left.clickCount == right.clickCount) {
          if (left.orderCount > right.orderCount) {
            true
          } else if (left.orderCount == right.orderCount) {
            left.payCount > right.payCount
          } else {
            false
          }
        } else {
          false
        }
      }
    }.take(10)


    result.foreach(println)


/*
    //将结果保存到Mysql中
    val driver = "com.mysql.jdbc.Driver"
    val url = "jdbc:mysql://hadoop102:3306/sparkmall190311"
    val userName = "root"
    val passWd = "111111"

    Class.forName(driver)
    val connection: Connection = DriverManager.getConnection(url, userName, passWd)
    val sql = "insert into category_top10 ( taskId, category_id, click_count, order_count, pay_count ) values (?, ?, ?, ?, ?)"
    val statement: PreparedStatement = connection.prepareStatement(sql)

    top10Array.foreach{
      obj=>{
        statement.setObject(1, obj.taskId)
        statement.setObject(2, obj.categoryId)
        statement.setObject(3, obj.clickCount)
        statement.setObject(4, obj.orderCount)
        statement.setObject(5, obj.payCount)
        statement.executeUpdate()
      }
    }
    statement.close()
    connection.close()*/
    // 释放资源
    sc.stop()
  }
}


//自定义累加器
//category-click,100
//categoru-order,200
//category-pay,300
class CategoryAccumulator extends AccumulatorV2[String,mutable.HashMap[String,Long]] {
  var map = new mutable.HashMap[String,Long]()

  override def isZero: Boolean = {
    map.isEmpty
  }

  override def copy(): AccumulatorV2[String, mutable.HashMap[String, Long]] = {
    new CategoryAccumulator
  }

  override def reset(): Unit = {
    map.clear()
  }

  override def add(in: String): Unit = {
    map(in) = map.getOrElse(in,0L) + 1
  }

  override def merge(other: AccumulatorV2[String, mutable.HashMap[String, Long]]): Unit = {
    var map1 = map
    var map2 = other.value

    map = map1.foldLeft(map2) {
      (tempMap, kv) =>{
        val k: String = kv._1
        val v: Long = kv._2

        tempMap(k) = tempMap.getOrElse(k,0L) + v

        tempMap
      }
    }
  }

  override def value: mutable.HashMap[String, Long] = {
    map
  }
}

4.2 需求1实现
一、实现二、使用累加器进行优化
需求实现1
实时统计注册人员信息用户使用网站或APP进行注册，后台实时收集数据传输Kafka，Spark Streaming进...
产品经理，你应该知道什么是需求分析
目录什么是产品什么是需求产品和需求的关系如何做需求分析4.1 从底层挖掘需求4.2 如何区分需求以及伪需求...
Swift基于GCD实现的倒计时功能
开发环境 1.编译器版本：10.12.sdk版本 12.13.swift 版本 4.2 实现代码
第 4 章需求工程
简介 4.1 功能型需求和非功能性需求 4.1.1 功能性需求 4.1.2 非功能性需求 4.2 需求工程过程 4...
理论知识
1.生理需求、安全需求、社交需求、尊重需求和自我实现需求。
（二）父组件传递事件给子组件（通过localStorage方式）
1、需求重点具体实现
需求决定的价格~4.2
人们很多情况下，高估了价格，而低估了价值，我想是因为我们根本就不清楚决定价格的从来不是约定俗成的成本价，而是你怎么...
python 实现迭代协议
4.4 如何实现一个比4.2节简单的迭代方法
centos7.2安装node.js环境
1,安装环境的要求 GCC 4.2 or newer; G++ 4.2 or newer; Python 2.6 ...