1.数据源

• 利用Python代码模拟生成，生成的数据如下：

     * id|brand|product|category|cost|price|quantity|province|timestamp
     * ---------------------------------------------------------------------------
     * 1|金龙鱼|金龙鱼花生油|生活用品|52|75|64|云南|1593536348
     * 2|金龙鱼|金龙鱼菜籽油|生活用品|48|62|16|河北|1593536348
     * 3|金龙鱼|金龙鱼玉米油|生活用品|34|53|72|河北|1593536348

• 生成数据的部分代码如下, 如果你需要完整代码，请关注我并私信"销售订单"

def produce_orders(file_name, num):
    provinces = (
         "江苏", "浙江", "安徽", "江西", "山东", 
         "山西", "河南", "河北", "湖北", "湖南", 
         "广东", "广西", "陕西", "甘肃", "福建", 
         "四川", "黑龙将", "辽宁",
         "吉林", "北京", "上海", "天津", "重庆", 
         "海南", "新疆", "云南", "贵州", "宁夏", "西藏", 
         "香港", "澳门", "台湾", "内蒙古")
    products = (
        ("金龙鱼花生油", "生活用品", 52, 75, "金龙鱼"), 
        ("金龙鱼植物油", "生活用品", 58, 80, "金龙鱼"),
        ("金龙鱼大豆油", "生活用品", 54, 73, "金龙鱼"),
...........
...........
...........
        ("西湖龙井", "茶叶", 22, 49, "龙井"), ("信阳毛尖", "茶叶", 13, 25, "信阳毛尖"),
        ("溧阳白茶", "茶叶", 23, 59, "溧阳白茶"), ("太平猴魁", "茶叶", 30, 69, "太平猴魁"), 
        ("安溪铁观音", "茶叶", 26, 43, "铁观音"),
        ("黄山毛峰", "茶叶", 15, 24, "毛峰")
    )
    with open(file_name, mode='a', encoding='utf-8') as f:
        counter = 0
        for i in range(num):
            for j in range(len(products)):
                counter += 1
                province = provinces[random.randint(0, len(provinces) - 1)]
                product_name = products[j][0]
                category = products[j][1]
                cost = products[j][2]
                price = products[j][3]
                brand = products[j][4]
                quantity = random.randint(10, 100)
                timestamp = int(time.time())

                record = str(counter) + "|" + brand + "|" + product_name + "|" + str(category) + "|"+ str(cost) + "|"+ str(price) + "|" + str(
                    quantity) + "|" + province + "|" + str(timestamp)
                f.write(record + "\n")
                time.sleep(0.0001)

• 数据集生成好了利用python代码自动上传HDFS

def write_to_hdfs(file_name):
    # 利用pyhdfs连接HDFS集群，用于上传生成的数据
    fs = pyhdfs.HdfsClient(hosts='node01:9870,node02:9870', randomize_hosts=False, user_name='hadoop', timeout=1000,
                           max_tries=1000)
    # HDFS要存放数据的目录
    hdfs_file_dir = '/user/hadoop/datasets/orders/'

    # 如果该目录不存在，则创建
    if not fs.exists(hdfs_file_dir):
        fs.mkdirs(hdfs_file_dir)

    # 上传文件, src为本地生成的文件路径，dest为hdfs上要存放的路径
    src = os.getcwd() + "/" + file_name
    dest = hdfs_file_dir + file_name
    # 从本地上传文件到hdfs指定目录
    fs.copy_from_local(src, dest)

2.Spark销售数据分析

由于目前我的文章只更新到了SparkCore部分，所以，现在的分析都是基于SparkCore提供的算子进行分析，在更新完SparkSql和SparkStreaming后会有更加方便的方法来分析这部分数据，后面会慢慢更新这部分内容。

为了清楚地表达代码的逻辑，分析这部分我使用了scala语言，后续也会整合python版本和java版本的。

• 计算销售额排名的前10的省份

  /**
   * 1.计算销售额排名的前10的省份
   *
   * @param productOrderCached
   */
  def computeTop10ProvinceSales(productOrderCached: RDD[ProductOrder]): Unit = {
    //把province和grossProfit组成K,V格式的RDD
    val provinceAndGrossProfit: RDD[(String, Float)] = productOrderCached.map(po => {
      (po.province, po.grossProfit)
    })

    //按照省份进行聚合
    val reducedProvinceAndGrossProfit: RDD[(String, Float)] = provinceAndGrossProfit.reduceByKey(_ + _)

    //按照聚合好的grossProfit进行倒排序
    val sortedProvinceAndGrossProfit: RDD[(String, Float)] = reducedProvinceAndGrossProfit.sortBy(tup => tup._2, false)

    //取出grossProfit前10省份的销量
    val results: Array[(String, Float)] = sortedProvinceAndGrossProfit.take(10)
    //打印结果，一般是生成json格式文件或者直接进入下游（比如，hdfs, mysql, kafka等)
    results.foreach(println)
  }

• 计算销售额排名的前5的种类

  /**
   * 2.计算销售额排名的前5的种类
   */
  def computeTop5CategorySales(productOrderCached: RDD[ProductOrder]) = {
    //把category和grossProfit组成K,V格式的RDD
    val categoryAndGrossProfit: RDD[(String, Float)] = productOrderCached.map(po => {
      (po.category, po.grossProfit)
    })

    //按照category进行聚合
    val reducedCategoryAndGrossProfit: RDD[(String, Float)] = categoryAndGrossProfit.reduceByKey(_ + _)

    //按照聚合好的grossProfit进行倒排序
    val sortedCategoryAndGrossProfit: RDD[(String, Float)] = reducedCategoryAndGrossProfit.sortBy(tup => tup._2, false)

    //取出grossProfit前10的category的销量
    val results: Array[(String, Float)] = sortedCategoryAndGrossProfit.take(5)
    //打印结果，一般是生成json格式文件或者直接进入下游（比如，hdfs, mysql, kafka等)
    results.foreach(println)
  }

• 计算最畅销的5个品牌

/**
   * 3.计算最畅销的5个品牌
   */
  def computeTop5Brand(productOrderCached: RDD[ProductOrder]) = {
    //对每条销售订单的brand计数1,并组成K,V格式的RDD
    val brandAndOne: RDD[(String, Int)] = productOrderCached.map(po => {
      (po.brand, 1)
    })

    //按照brand进行聚合
    val reducedBrandCounts: RDD[(String, Int)] = brandAndOne.reduceByKey(_ + _)

    //按照brand的计数进行倒排序
    val sortedBrandAndCounts: RDD[(String, Int)] = reducedBrandCounts.sortBy(tup => tup._2, false)

    //取出前10
    val results: Array[(String, Int)] = sortedBrandAndCounts.take(5)

    results.foreach(println)
  }

• 统计最挣钱的10个商品

  /**
   * 4.统计最挣钱的10个商品
   */
  def computeTop10NetProfit(productOrderCached: RDD[ProductOrder]) = {
    // 取出产品名称和净利润组成K,V格式的RDD
    val productAndNetProfit: RDD[(String, Float)] = productOrderCached.map(po => {
      (po.product, po.netProfit)
    })

    // 按产品聚合
    val reducedProductAndNetProfit: RDD[(String, Float)] = productAndNetProfit.reduceByKey(_ + _)
    // 按照净利润倒排序
    val sortedProductAndNetProfit: RDD[(String, Float)] = reducedProductAndNetProfit.sortBy(tup => tup._2, false)

    // 取出前10
    val results: Array[(String, Float)] = sortedProductAndNetProfit.take(10)

    //打印结果
    results.foreach(println)

  }

后话

以上只是利用现有的SparkCore做的销售数据的分析，后面会持续更新利用SparkSQL+Hive与SparkStreaming对接Kafka用流处理的方式来分析处理数据。

由于测试数据文件比较大(建议下载源码生成)，如果您需要全部源码与生成的测试数据，请关注我并发送私信"销售订单"。