ansible 一些基本使用方法的记录

本文可看做是 用户行为分析(pyspark) 的后续篇, 主要介绍使用spark统计各区域热门商品,完整代码可以点击参考这里,代码中有详细的注释,所以可以也可以略过文章,直接看代码项目源码地址

模拟数据

模拟数据与 用户行为分析(pyspark) 中的模拟数据一样

需求

根据用户指定的日期范围,统计各个区域下最热门的3个产品

方案

产品的热度根据产品的点击量评价

1. 根据筛选条件,查询用户行为记录表中的区域id和点击的产品id

sql_str = "SELECT city_id, click_product_id as product_id  
           FROM user_visit_action 
           WHERE click_product_id != '' AND  date >='" + start_date + "' AND date<='" + end_date + "'"

2. 获取城市信息

# 实际应用中,这一步可能是从关系型数据库中获取城市信息,这里为了方便测试,模拟一份数据
def get_caty_id_to_caty_info_rdd(sc):
    """
    获取城市详细信息
    :param sc:
    :return:
    """
    # 城市详细信息(id,(id,name,area))
    caty_info = [
        (0, (0, '北京', '华北')),
        (1, (1, '上海', '华东')),
        (2, (2, '南京', '华东')),
        (3, (3, '广州', '华南')),
        (4, (4, '三亚', '华南')),
        (5, (5, '武汉', '华中')),
        (6, (6, '长沙', '华中')),
        (7, (7, '西安', '西北')),
        (8, (8, '成都', '西南')),
        (9, (9, '哈尔滨', '东北'))

    ]
    return sc.parallelize(caty_info)

3. 创建包含 city_id, city_name, area, product_id 的临时表 tmp_clk_prod_basie

def get_temp_click_product_table(spark, click_action_rdd, city_id_to_city_info_rdd):
    """
    把 rdd 合并得到每次action的city_id和product_id, 并转化为临时表
    :param spark:
    :param click_action_rdd:
    :param city_id_to_city_info_rdd:
    :return:
    """

    temp_click_product_rdd = click_action_rdd\
        .join(city_id_to_city_info_rdd)\
        .map(lambda row: Row(city_id=row[0], city_name=row[1][1][1], area=row[1][1][2], product_id=row[1][0][1]))

    schema_click_product = spark.createDataFrame(temp_click_product_rdd)
    schema_click_product.createOrReplaceTempView("tmp_clk_prod_basic")
    # schema_click_product.show()

4. 按照 area, product_id 分组, 计算 click_count 创建 tmp_area_product_click_count

def schema_temp_area_prdocut_click_count_table(spark):
    """
    生成各区域各商品点击次数临时表
    :param spark:
    :return:
    """
    sql = "SELECT " \
          "area, " \
          "product_id, " \
          "count(*) as click_count, " \
          "collect_set(concat(city_id, ':', city_name)) as city_info " \
          "FROM tmp_clk_prod_basic " \
          "GROUP BY area, product_id"
    df = spark.sql(sql)
    df.createOrReplaceTempView("tmp_area_product_click_count")

5. 利用开窗函数从 tmp_area_product_click_count 取出每个区域中点击量排名前三的 product_id

def get_area_top3_product_rdd(spark):
    """
    获取各区域 top3 商品
    先使用开窗函数进行一个子查询, 查询出一个 每个区域内 商品的 click_count 按照倒序排序的结果集
    然后在得到的结果集中,取每个区域中 行号 1-3 的行
    :param spark:
    :return:
    """
    sql = "SELECT " \
            "area, " \
            "CASE " \
                "WHEN area='华北' OR area='华东' THEN 'A级' " \
                "WHEN area='华中' OR area='华南'  THEN 'B级' " \
                "WHEN area='西北' OR area='西南'  THEN 'C级' " \
                "ELSE 'D级'" \
            "END area_level, " \
          "product_id, " \
          "click_count, " \
          "city_info " \
          "FROM " \
            "(" \
                "SELECT " \
                    "area, " \
                    "product_id, " \
                    "click_count, " \
                    "city_info, " \
                    "row_number() OVER(PARTITION BY area ORDER BY click_count DESC ) as rank " \
                    "FROM tmp_area_product_click_count) AS t " \
          "WHERE rank <=3"

    df = spark.sql(sql)
    df.show()