day15_挖掘类标签和推荐系统

一、基于决策树的USG标签开发（熟悉）

1、计算方案

在统计类标签开发中，根据用户购买男性特征商品和女性特征商品的数量和比例来确定用户的购物性别。这种方式属于硬编码，虽然解释性强，但因为考虑的因素比较少，所以准确率会比较低。

而决策树的优点较多，主要是其变量处理灵活，不要求相互独立。可处理大维度的数据，不用预先对模型的特征有所了解。对于表达复杂的非线性模式和特征的相互关系，模型相对容易理解和解释。看起来决策树的方法最适合区分性别特征了，所以决定用决策树进行尝试。

2、代码实现

• 决策树算法：

from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.feature import VectorAssembler
from pyspark.sql import DataFrame
import pyspark.sql.functions as F
from pyspark.sql.types import StringType

from tags.base.abstract_tag_base import AbstractTagBase

class UsgTagML(AbstractTagBase):
    def get_order_gender(self, tmp_row):
        # zt_id,order_no,third_category_no,goods_no,order_total_amount,order_discount_amount
        # 1- 定义临时变量
        # 男性标识，1标识男性购物者，0表示非男性购物者
        male_type = 0
        # 女性标识，1标识女性购物者，0表示非女性购物者
        female_type = 0

        third_category_no = tmp_row.third_category_no
        goods_no = tmp_row.goods_no

        # 判断是否是男性购物者
        if third_category_no in ("20020105","20020101","20020303","20020304","20020201","20020801") \
            or goods_no in ("2904158","3224617","3229811","3231330","3222118"):
            male_type = 1

        # 判断是否是女性购物者
        if third_category_no in ("20020102","20020402","10050507","10020403","30050301","30050401","30020901") \
                or goods_no in ("0301725","3221471","3224069","3225168","3232149","3216901"):
            female_type = 1

        return tmp_row.zt_id,male_type,female_type,tmp_row.order_no,tmp_row.third_category_no,tmp_row.goods_no,tmp_row.order_total_amount,tmp_row.order_discount_amount


    def mark_tag(self,business_df:DataFrame, five_tag_df:DataFrame):
        # 1- 区分每笔订单是什么性别
        # 这里对小数需要使用decimal类型进行接收
        gender_type_df = business_df.rdd.map(self.get_order_gender).toDF("zt_id:int,male_type:int,female_type:int,order_no:string,third_category_no:string,goods_no:string,order_total_amount:decimal,order_discount_amount:decimal")

        # 2- 指标统计
        """
            男性商品数占比 = 男性订单数 / 总订单数
            女性商品数占比 = 女性订单数 / 总订单数
            平均每人买的商品种类数 = 商品种类数 / 总订单数
            平均每人消费金额 = 总金额 / 总订单数
            平均每人折扣金额 = 总的折扣金额 / 总的折扣订单数
            折扣单量占比 = 总的折扣订单数 / 总订单数
            折扣金额占比 = 总的折扣金额 / 总金额
        """
        # 分别计算分子和分母
        # zt_id,male_type,female_type,order_no,third_category_no,goods_no,order_total_amount,order_discount_amount
        tmp_business_df = gender_type_df.groupBy("zt_id").agg(
            F.sum("male_type").alias("male_count"), # 男性订单数
            F.sum("female_type").alias("female_count"), # 女性订单数
            F.countDistinct("third_category_no").alias("category_count"), # 商品种类数
            F.sum("order_discount_amount").alias("discount_amount"), # 总的折扣金额
            F.sum(F.when(gender_type_df.order_discount_amount>0,1).otherwise(0)).alias("discount_count"), # 总的折扣订单数
            F.sum("order_total_amount").alias("total_amount"), # 总金额
            F.countDistinct("order_no").alias("total_order_count"), # 总订单数
        )

        # 计算最终的指标
        new_business_df = tmp_business_df.select(
            "zt_id",
            (tmp_business_df.male_count/tmp_business_df.total_order_count).alias("male_rate"), # 男性商品数占比
            (tmp_business_df.female_count/tmp_business_df.total_order_count).alias("female_rate"), # 女性商品数占比
            (tmp_business_df.category_count/tmp_business_df.total_order_count).alias("category_num"), # 平均每人买的商品种类数
            (tmp_business_df.total_amount/tmp_business_df.total_order_count).alias("total_money"), # 平均每人消费金额
            (tmp_business_df.discount_amount/tmp_business_df.discount_count).alias("discount_money"), # 平均每人折扣金额
            (tmp_business_df.discount_count/tmp_business_df.total_order_count).alias("discount_rate"), # 折扣单量占比
            (tmp_business_df.discount_amount/tmp_business_df.total_amount).alias("discount_money_rate") # 折扣金额占比
        )

        # 3- 与用户注册表进行关联，打上标签
        """
            1- 因为new_business_df数据中没有用户性别信息，因此需要额外的查询用户注册表dwd.dwd_mem_member_union_i的zt_id、sex字段
            2- 但是当前方法mark_tag中并没有将用户注册表的数据传递进来，所以需要自己通过SparkSession对象单独的读取这张表的数据
            3- 读取Hive表的数据可以通过调用父类中的read_hive_data方法实现，但是它里面需要传递进SparkSession对象
            4- 接下来要想办法得到SparkSession对象
                4.1- 调整父类的create_spark方法，将里面的spark变量前面带上self
                4.2- 在父类中新创建一个获得SparkSession对象的实例方法
        """
        # 我们虽然这么分析了，但是不能这么用，因为SparkSession/SparkContext只能在Driver进程中创建使用https://blog.csdn.net/qwer123451234123/article/details/124479205
        # spark = self.get_spark()
        # all_tag_info_df = self.read_all_tag(spark)
        # rule_obj = self.read_and_parse_rule(all_tag_info_df, four_tag_id=4)
        # user_info_df = self.read_hive_data(rule_obj, spark, condition=None)

        spark = self.create_spark("sex_tag",2)
        user_info_df = spark.sql("select zt_id,sex from dwd.dwd_mem_member_union_i")

        # 如果join的两个DF中对应的关联字段相同，可以直接传一个字段名即可
        final_business_df = new_business_df.join(user_info_df,"zt_id",how="left").select(
            new_business_df.zt_id,
            user_info_df.sex,
            new_business_df.male_rate,
            new_business_df.female_rate,
            new_business_df.category_num,
            new_business_df.total_money,
            new_business_df.discount_money,
            new_business_df.discount_rate,
            new_business_df.discount_money_rate
        ).fillna(0)

        # final_business_df.show()
        # final_business_df.printSchema()

        # 4- 特征预处理
        # 字段处理
        assembler = VectorAssembler(
            inputCols=["male_rate","female_rate","category_num","total_money","discount_money","discount_rate","discount_money_rate"],
            outputCol="features"
        )
        assembler_df = assembler.transform(final_business_df)

        # 数据集划分
        train_df,test_df = assembler_df.randomSplit(weights=[0.8,0.2])

        # 5- 决策树模型
        # 5.1- 实例化决策树模型
        model = DecisionTreeClassifier(
            featuresCol="features",
            labelCol="sex",
            predictionCol="prediction"
        )

        # 5.2- 对模型进行训练
        tree_model = model.fit(train_df)

        # 5.3- 处理数据
        tree_result_df = tree_model.transform(assembler_df)
        tree_result_df.show()
        tree_result_df.printSchema()

        # 6- 打标签
        five_tag_dict = five_tag_df.rdd.map(lambda row:(row.rule, row.id)).collectAsMap()
        print(five_tag_dict)

        # 注意：prediction的数据类型是double，而five_tag_dict的key值是字符串内容是整数。因此需要进行类型转换
        @F.udf(returnType=StringType())
        def get_tag_id(prediction):
            return five_tag_dict.get(str(int(prediction)))

        result_df = tree_result_df.select(
            tree_result_df.zt_id.alias("user_id"),
            get_tag_id(tree_result_df.prediction).alias("tags_id_times")
        )
        result_df.show(n=10000)
        return result_df

if __name__ == '__main__':
    condition = " and datediff(`current_date`(),to_date(trade_date))<=90 and zt_id!=0 and zt_id is not null "

    tag_obj = UsgTagML()
    tag_obj.execute(
        app_name="usg_tag_ml",
        partitions=2,
        four_tag_id=58,
        condition=condition
    )

• 也可以使用KMeans的聚类来实现

import numpy as np
from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.clustering import KMeans
from pyspark.ml.feature import VectorAssembler
from pyspark.sql import DataFrame
import pyspark.sql.functions as F
from pyspark.sql.types import StringType

from tags.base.abstract_tag_base import AbstractTagBase

class UsgTagML(AbstractTagBase):
    def get_order_gender(self, tmp_row):
        # zt_id,order_no,third_category_no,goods_no,order_total_amount,order_discount_amount
        # 1- 定义临时变量
        # 男性标识，1标识男性购物者，0表示非男性购物者
        male_type = 0
        # 女性标识，1标识女性购物者，0表示非女性购物者
        female_type = 0

        third_category_no = tmp_row.third_category_no
        goods_no = tmp_row.goods_no

        # 判断是否是男性购物者
        if third_category_no in ("20020105","20020101","20020303","20020304","20020201","20020801") \
            or goods_no in ("2904158","3224617","3229811","3231330","3222118"):
            male_type = 1

        # 判断是否是女性购物者
        if third_category_no in ("20020102","20020402","10050507","10020403","30050301","30050401","30020901") \
                or goods_no in ("0301725","3221471","3224069","3225168","3232149","3216901"):
            female_type = 1

        return tmp_row.zt_id,male_type,female_type,tmp_row.order_no,tmp_row.third_category_no,tmp_row.goods_no,tmp_row.order_total_amount,tmp_row.order_discount_amount


    def mark_tag(self,business_df:DataFrame, five_tag_df:DataFrame):
        # 1- 区分每笔订单是什么性别
        # 这里对小数需要使用decimal类型进行接收
        gender_type_df = business_df.rdd.map(self.get_order_gender).toDF("zt_id:int,male_type:int,female_type:int,order_no:string,third_category_no:string,goods_no:string,order_total_amount:decimal,order_discount_amount:decimal")

        # 2- 指标统计
        """
            男性商品数占比 = 男性订单数 / 总订单数
            女性商品数占比 = 女性订单数 / 总订单数
            平均每人买的商品种类数 = 商品种类数 / 总订单数
            平均每人消费金额 = 总金额 / 总订单数
            平均每人折扣金额 = 总的折扣金额 / 总的折扣订单数
            折扣单量占比 = 总的折扣订单数 / 总订单数
            折扣金额占比 = 总的折扣金额 / 总金额
        """
        # 分别计算分子和分母
        # zt_id,male_type,female_type,order_no,third_category_no,goods_no,order_total_amount,order_discount_amount
        tmp_business_df = gender_type_df.groupBy("zt_id").agg(
            F.sum("male_type").alias("male_count"), # 男性订单数
            F.sum("female_type").alias("female_count"), # 女性订单数
            F.countDistinct("third_category_no").alias("category_count"), # 商品种类数
            F.sum("order_discount_amount").alias("discount_amount"), # 总的折扣金额
            F.sum(F.when(gender_type_df.order_discount_amount>0,1).otherwise(0)).alias("discount_count"), # 总的折扣订单数
            F.sum("order_total_amount").alias("total_amount"), # 总金额
            F.countDistinct("order_no").alias("total_order_count"), # 总订单数
        )

        # 计算最终的指标
        final_business_df = tmp_business_df.select(
            "zt_id",
            (tmp_business_df.male_count/tmp_business_df.total_order_count).alias("male_rate"), # 男性商品数占比
            (tmp_business_df.female_count/tmp_business_df.total_order_count).alias("female_rate"), # 女性商品数占比
            (tmp_business_df.category_count/tmp_business_df.total_order_count).alias("category_num"), # 平均每人买的商品种类数
            (tmp_business_df.total_amount/tmp_business_df.total_order_count).alias("total_money"), # 平均每人消费金额
            (tmp_business_df.discount_amount/tmp_business_df.discount_count).alias("discount_money"), # 平均每人折扣金额
            (tmp_business_df.discount_count/tmp_business_df.total_order_count).alias("discount_rate"), # 折扣单量占比
            (tmp_business_df.discount_amount/tmp_business_df.total_amount).alias("discount_money_rate") # 折扣金额占比
        ).fillna(0)

        # 3- 使用KMeans实现聚类
        # 3.1- 特征预处理
        assembler = VectorAssembler(
            inputCols=["male_rate","female_rate","category_num","total_money","discount_money","discount_rate","discount_money_rate"],
            outputCol="features"
        )

        # 对数据内容进行真正的处理
        vector_business_df = assembler.transform(final_business_df)

        # 3.2- 实例化KMeans算法模型
        """
            1- featuresCol的参数值需要与前面VectorAssembler设置的outputCol的值保持一致
            2- 本次k的值，需要与对应的五级标签个数相同
            3- seed的值没有固定大小，随意
            4- predictionCol算法模型预测结果值存放的字段
        """
        kmeans = KMeans(featuresCol="features",predictionCol="prediction",k=3,seed=1)

        # 3.3- 模型训练
        kmeans_model = kmeans.fit(vector_business_df)

        # 3.4- 训练好的模型对业务数据进行处理
        kmeans_result:DataFrame = kmeans_model.transform(vector_business_df)

        # 4- 给用户打上具体的标签
        # 4.1- 获取质心位置。有7个质心，每个质心是三维的
        cluster_centers_list = kmeans_model.clusterCenters()

        # 4.2- 对质心进行求和
        center_list_sum = [np.sum(center) for center in cluster_centers_list]

        # 4.3- 构造一个质心得分字典
        center_dict_sum = {}
        for i in range(len(center_list_sum)):
            center_dict_sum[i] = center_list_sum[i]

        # 4.4- 将字典的value进行降序排序
        """
            sorted：对容器进行排序，默认是升序
            key：自定义排序规则
            reverse：是否对排完序以后的内容进行顺序反转。默认不反转
        """
        ordered_dict = dict(sorted(center_dict_sum.items(),key=lambda item:item[1],reverse=True))

        # 4.5- 将排完序的聚合结果与五级标签配置数据关联起来
        five_tag_id_list = five_tag_df.rdd.map(lambda row:str(row.id)).collect()
        kmeans_and_tagid_dict = dict(zip(ordered_dict.keys(),five_tag_id_list))

        # 4.6- 真正的打标签
        @F.udf(returnType=StringType())
        def get_tag_id(prediction_id):
            return kmeans_and_tagid_dict.get(prediction_id)

        result_df = kmeans_result.select(
            kmeans_result.zt_id.alias("user_id"),
            get_tag_id(kmeans_result.prediction).alias("tags_id_times")
        )

        return result_df

if __name__ == '__main__':
    condition = " and datediff(`current_date`(),to_date(trade_date))<=90 and zt_id!=0 and zt_id is not null "

    tag_obj = UsgTagML()
    tag_obj.execute(
        app_name="usg_tag_ml",
        partitions=2,
        four_tag_id=58,
        condition=condition
    )

可能遇到的错误：

原因: 是因为数据中有空值
解决: 针对业务数据使用fillna填充

二、SparkML的模型选择和优化（了解）

1、模型选择

ML中的一个重要任务是模型选择，或使用数据找到给定任务的最佳模型或参数。这也叫调优。可以针对个体估算器（如Logistic回归）或包括多个算法，特征化和其他步骤的整个管道完成调整。用户可以一次调整整个流水线，而不是单独调整管道中的每个元素。

MLlib支持使用 CrossValidator和TrainValidationSplit等工具进行模型选择。

实际工作中如何选择无监督和有监督的算法呢？
1- 项目/网站/App等的初期，因为没有太多的历史数据，因此一般选择无监督算法
2- 如果你的项目/网站/App等已经运行了很长一段时间，那么数据积累是比较多的，有干净准确的数据，因此一般后期使用有监督算法

2、交叉验证

CrossValidator首先将数据集分成一组折叠，这些折叠用作单独的训练和测试数据集。 例如，k = 3倍，CrossValidator将生成3个（训练，测试）数据集对，每个数据集使用2/3的数据进行训练，1/3进行测试。为了评估一个特定的ParamMap，CrossValidator通过在3个不同的（训练，测试）数据集对上拟合Estimator来计算3个模型的平均评估度量。在确定最佳ParamMap之后，CrossValidator最终使用最好的ParamMap和整个数据集重新拟合Estimator。

3、训练验证方法切分

TrainValidationSplit仅对参数每个组合进行一次评估，而在CrossValidator的情况下，则不是k次。 因此，它较便宜，但在训练数据集不够大时不会产生可靠的结果。与CrossValidator不同，TrainValidationSplit创建一个（训练，测试）数据集对。 它使用trainRatio参数将数据集分成这两个部分。 例如，trainRatio = 0.75，TrainValidationSplit将生成训练和测试数据集对，其中75％的数据用于训练，25％用于验证。

像CrossValidator一样，TrainValidationSplit最终适合使用最好的ParamMap和整个数据集的Estimator。

4、Pipeline构建模型

使用数据集，构建决策树分类模型，依据购买的商品和颜色预测其性别，为后续判断用户购物性别做依据。由于需要对数据集进行特征工程，所以使用Pipeline管道进行封装，获取管道模型PipelineModel。

5、模型调优

使用决策树算法训练模型时，可以调整相关参数，结合训练验证或交叉验证，获取最佳模型。

三、推荐系统介绍

推荐系统一般是由Java后端与前端人员进行开发的，大数据开发人员比较少参与主要是提供数据。

1、什么是推荐系统

为了解决信息过载和用户无明确需求的问题，找到用户感兴趣的物品，才有了个性化推荐系统。

2、推荐模块概述

推荐的方式主要分为两种：离线推荐和实时推荐。其中实时推荐一般是离线跑好模型，然后将模型进行部署，根据实时采集的数据实时进行推荐。离线推荐模块则是需要综合用户所有的历史数据，利用设定的离线统计算法和离线推荐算法周期性的进行结果统计与保存，一般运行时间比较长。

推荐算法主要包括基于流行度推荐、基于隐语义模型的协同过滤推荐、基于物品的协同过滤推荐、基于用户的协同过滤推荐以及基于关联规则的推荐。

四、推荐实现

为了实现推荐功能，需要启动Hadoop、Hive、ES、Doris、SparkSubmit
启动Hadoop、启动Hive
    cd /
    ./up01.sh start

启动ES
    1- 切换用户
    su es

    2- 进入目录
    cd /home/es/elasticsearch-7.10.2/bin

    3- 启动
    elasticsearch -d
    
    4- 退出es用户
    exit

启动Doris
    /export/server/doris/fe/bin/start_fe.sh --daemon
    /export/server/doris/be/bin/start_be.sh --daemon
    /export/server/doris/apache_hdfs_broker/bin/start_broker.sh --daemon

启动SparkSubmit
    cd /export/server/spark/sbin

    ./start-thriftserver.sh \
    --hiveconf hive.server2.thrift.port=10001 \
    --hiveconf hive.server2.thrift.bind.host=up01 \
    --hiveconf spark.sql.warehouse.dir=hdfs://up01:8020/user/hive/warehouse \
    --master local[*]

1、基于流行度推荐（掌握）

基于流行度推荐，也就是基于统计的推荐，主要用来解决用户的冷启动问题，对于新用户首次登录。基于流行度的推荐也可以用于单独的热门商品模块。

1.1 近期热门商品推荐

可以按商品销售的单量进行倒序排序，然后存入Doirs中。表中可以多存入一些数据，在使用时，根据品类进行倒序查询，取到相关商品即可。

• 计算的sql如下

-- 近期热门商品推荐
select
    current_date() as recommend_date, -- 推荐时间，区别是什么时候推荐的
    third_category_no,
    third_category_name,
    goods_no,
    goods_name,
    count(order_no) as order_count -- 订单量
from dwm.dwm_sold_goods_sold_dtl_i
where
    -- 过滤最近一段时间内的销售数据
    datediff(current_date(),to_date(trade_date))<=40 and goods_no is not null
group by third_category_no,third_category_name,goods_no,goods_name
order by order_count desc
limit 300 -- 推荐比项目经理要求的推荐数目多一些

• Doris的建表语句如下

create database if not exists recommend_db;
CREATE TABLE IF NOT EXISTS recommend_db.popular_hot_goods (
    recommend_date DATE comment '计算日期',
    goods_no bigint comment '商品编码',
    third_category_no STRING comment '三级品类编码',
    third_category_name STRING comment '三级品类名称',
    goods_name STRING comment '商品名称',
    order_count INT comment '销售数量'
)
UNIQUE KEY(recommend_date, goods_no)
comment '热门商品推荐'
PARTITION BY RANGE(recommend_date) ()
DISTRIBUTED BY HASH(goods_no) BUCKETS 1
PROPERTIES (
    "dynamic_partition.create_history_partition" = "true",
    "dynamic_partition.enable" = "true",
    "dynamic_partition.time_unit" = "DAY",
    "dynamic_partition.start" = "-365",
    "dynamic_partition.end" = "3",
    "dynamic_partition.prefix" = "p",
    "dynamic_partition.buckets" = "10",
    "replication_allocation" = "tag.location.default: 1"
);

注意: 
	1- 如果多个字段作为UNIQUE KEY，那么String类型不能够使用。因此这里将goods_no的类型进行了强制转换。为了将不同日期的数据分开存放，这里使用动态分区表。
	2- 建表中的字段顺序需要与UNIQUE KEY中字段顺序保持一致。并且UNIQUE KEY中的字段要放在最上面。如果不遵守会报如下的错：

• 推荐代码

from pyspark.sql import SparkSession
import os
import pyspark.sql.functions as F

os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

if __name__ == '__main__':
    # 1- 创建SparkSession对象
    spark = SparkSession.builder\
        .appName("topn_goods_recommend")\
        .master("local[*]") \
        .config("spark.sql.warehouse.dir", "hdfs://up01:8020/user/hive/warehouse") \
        .config("hive.metastore.uris", "thrift://up01:9083") \
        .config("spark.sql.shuffle.partitions",2)\
        .enableHiveSupport() \
        .getOrCreate()

    # 2- 读取分析Hive中的数据，获取TOPN的热门商品
    topn_goods_df = spark.sql("""
        select
            current_date() as recommend_date, -- 推荐时间，区别是什么时候推荐的
            third_category_no,
            third_category_name,
            goods_no,
            goods_name,
            count(order_no) as order_count -- 订单量
        from dwm.dwm_sold_goods_sold_dtl_i
        where
            -- 过滤最近一段时间内的销售数据
            datediff(current_date(),to_date(trade_date))<=40 and goods_no is not null
        group by third_category_no,third_category_name,goods_no,goods_name
        order by order_count desc
        limit 300 -- 推荐比项目经理要求的推荐数目多一些
    """)

    # 3- 数据存储到Doris中
    topn_goods_df.write.jdbc(
        url="jdbc:mysql://192.168.88.166:9030/recommend_db?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&useSSL=false",
        table="popular_hot_goods",
        mode="append",
        properties={ 'user' : 'root', 'password' : '123456' }
    )

    # 4- 释放资源
    spark.stop()

1.2 个人热门商品推荐

也就是根据销售单量统计每个人喜欢购买的前N个商品

• 计算的sql如下

select
    current_date() as recommend_date, -- 推荐时间，区别是什么时候推荐的
    zt_id as user_id,
    goods_no,
    goods_name,
    third_category_no,
    third_category_name,
    order_count
from (
    select
        *,
        row_number() over(partition by zt_id order by order_count desc) as rn
    from (
        select
            zt_id,
            third_category_no,
            third_category_name,
            goods_no,
            goods_name,
            count(order_no) as order_count -- 订单量
        from dwm.dwm_sold_goods_sold_dtl_i
        where
            -- 过滤最近一段时间内的销售数据
            datediff(current_date(),to_date(trade_date))<=40 and goods_no is not null
        and zt_id is not null and zt_id!=0
        group by zt_id,third_category_no,third_category_name,goods_no,goods_name
    ) tmp_1
) tmp_2 where rn<=20

• Doris建表语句

CREATE TABLE IF NOT EXISTS recommend_db.popular_person_hot_goods (
    recommend_date DATE comment '计算日期',
    user_id INT comment '会员ID',
    goods_no STRING comment '商品编码',
    goods_name STRING comment '商品名称',
    third_category_no STRING comment '三级品类编码',
    third_category_name STRING comment '三级品类名称',
    order_count INT comment '订单数'
)
UNIQUE KEY(recommend_date, user_id)
comment '个人热门商品推荐'
PARTITION BY RANGE(recommend_date) ()
DISTRIBUTED BY HASH(user_id) BUCKETS 1
PROPERTIES (
    "dynamic_partition.create_history_partition" = "true",
    "dynamic_partition.enable" = "true",
    "dynamic_partition.time_unit" = "DAY",
    "dynamic_partition.start" = "-365",
    "dynamic_partition.end" = "3",
    "dynamic_partition.prefix" = "p",
    "dynamic_partition.buckets" = "10",
    "replication_allocation" = "tag.location.default: 1"
);

• 推荐代码

from pyspark.sql import SparkSession
import os
import pyspark.sql.functions as F

os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

if __name__ == '__main__':
    # 1- 创建SparkSession对象
    spark = SparkSession.builder\
        .appName("topn_user_goods_recommend")\
        .master("local[*]") \
        .config("spark.sql.warehouse.dir", "hdfs://up01:8020/user/hive/warehouse") \
        .config("hive.metastore.uris", "thrift://up01:9083") \
        .config("spark.sql.shuffle.partitions",8)\
        .enableHiveSupport() \
        .getOrCreate()

    # 2- 读取分析Hive中的数据，获取TOPN的热门商品
    topn_goods_df = spark.sql("""
        select
            current_date() as recommend_date, -- 推荐时间，区别是什么时候推荐的
            zt_id as user_id,
            goods_no,
            goods_name,
            third_category_no,
            third_category_name,
            order_count
        from (
            select
                *,
                row_number() over(partition by zt_id order by order_count desc) as rn
            from (
                select
                    zt_id,
                    third_category_no,
                    third_category_name,
                    goods_no,
                    goods_name,
                    count(order_no) as order_count -- 订单量
                from dwm.dwm_sold_goods_sold_dtl_i
                where
                    -- 过滤最近一段时间内的销售数据
                    datediff(current_date(),to_date(trade_date))<=40 and goods_no is not null
                and zt_id is not null and zt_id!=0
                group by zt_id,third_category_no,third_category_name,goods_no,goods_name
            ) tmp_1
        ) tmp_2 where rn<=20
    """)

    # 3- 数据存储到Doris中
    topn_goods_df.write.jdbc(
        url="jdbc:mysql://192.168.88.166:9030/recommend_db?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&useSSL=false",
        table="popular_person_hot_goods",
        mode="append",
        properties={ 'user' : 'root', 'password' : '123456' }
    )

    # 4- 释放资源
    spark.stop()