dwd_feedback.py 16.7 KB
Newer Older
chenyuanjie committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355
"""
   @Author      : HuangJian
   @Description : 店铺feedBack页面top20相关指标统计
   @SourceTable :
                  ①ods_seller_account_syn
                  ②ods_seller_asin_account
                  ③ods_seller_account_feedback_report
                  ④ods_asin_detail_product
                  ⑤dim_asin_history_info
   @SinkTable   :
                  ①dwd_seller_asin_account_agg
                  ②dwd_seller_asin_account_detail
   @CreateTime  : 2022/11/2 9:56
   @UpdateTime  : 2022/11/2 9:56
"""

import datetime
import traceback
import os
import sys
from datetime import date, timedelta

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from pyspark.sql.functions import ceil
from pyspark.sql.types import IntegerType
from utils.templates import Templates
# from ..utils.templates import Templates
from pyspark.sql import functions as F
from yswg_utils.common_udf import udf_get_package_quantity


class DwdFeedBack(Templates):
    def __init__(self, site_name='us', date_type="week", date_info='2022-40'):
        super().__init__()
        self.db_save = "feedback_week"
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        self.spark = self.create_spark_object(
            app_name=f"{self.db_save}:{self.site_name}_{self.date_type}_{self.date_info}")
        self.df_date = self.get_year_week_tuple()
        self.month = self.get_month_by_week()
        self.week_date = self.get_calDay_by_dateInfo()
        self.month_old = int(self.month)
        self.ym = f"{self.year}_{self.month_old}"
        self.partitions_by = ['site_name', 'date_type', 'date_info']

        # 自定义udf函数相关对象
        self.u_launch_time = self.spark.udf.register("u_launch_time", self.udf_launch_time, IntegerType())
        self.u_days_diff = self.spark.udf.register("u_days_diff", self.udf_days_diff, IntegerType())
        self.u_judge_package_quantity = F.udf(udf_get_package_quantity,IntegerType())
        # 初始化全局变量df--ods获取数据的原始df
        self.df_seller_acount_syn = self.spark.sql("select 1+1;")
        self.df_seller_asin_acount = self.spark.sql("select 1+1;")
        self.df_seller_account_feedback_report = self.spark.sql("select 1+1;")
        self.df_asin_detail_product = self.spark.sql("select 1+1;")
        self.df_asin_detail = self.spark.sql("select 1+1;")
        # 初始化全局变量df--dwd层转换输出的df
        self.df_seller = self.spark.sql("select 1+1;")
        self.df_seller_agg = self.spark.sql("select 1+1;")
        self.df_seller_detail = self.spark.sql("select 1+1;")
        self.df_asin_parent = self.spark.sql("select 1+1;")

        # 初始化全局变量df -- 中间过程使用
        self.df_seller_top = self.spark.sql("select 1+1;")
        self.df_asin_counts = self.spark.sql("select 1+1;")
        self.df_asin_new_counts = self.spark.sql("select 1+1;")
        self.df_variat_ratio = self.spark.sql("select 1+1;")

    @staticmethod
    def udf_launch_time(launch_time, cal_day):
        # 针对launch_time字段进行计算与当前日期的间隔天数
        if "-" in str(launch_time):
            # print(DwdFeedBack.week_date)
            asin_date_list = str(launch_time).split("-")
            try:
                asin_date = datetime.date(year=int(asin_date_list[0]),
                                          month=int(asin_date_list[1]),
                                          day=int(asin_date_list[2]))
                if not cal_day.strip():
                    week_date = '2022-11-02'
                else:
                    week_date = cal_day
                cur_date_list = str(week_date).split("-")
                cur_date = datetime.date(year=int(cur_date_list[0]),
                                         month=int(cur_date_list[1]),
                                         day=int(cur_date_list[2]))
                days_diff = (cur_date - asin_date).days
            except Exception as e:
                print(e, traceback.format_exc())
                print(launch_time, asin_date_list)
                days_diff = 999999
        else:
            days_diff = 999999
        return days_diff

    @staticmethod
    def udf_days_diff(days_diff):
        # 针对days_diff字段进行计算180天,判断是否为新品
        if days_diff <= 180:
            return 1
        elif days_diff == 999999:
            return None
        else:
            return 0

    def get_month_by_week(self):
        if self.date_type == "week":
            df = self.df_date.loc[self.df_date.year_week == self.year_week]
            month = list(df.month)[0]
            if int(month) < 10:
                month = "0" + str(month)
            print("month:", month)
            return str(month)

    def get_calDay_by_dateInfo(self):
        if self.date_type in ['day', 'last30day']:
            return str(self.date_info)
        # 如果为 周、月则取该周、月的最后一日,作为新品计算基准日
        if self.date_type in ['week', 'month']:
            self.df_date = self.spark.sql(f"select * from dim_date_20_to_30;")
            df = self.df_date.toPandas()
            df_loc = df.loc[df[f'year_{self.date_type}'] == f"{self.date_info}"]
            self.date_info_tuple = tuple(df_loc.date)
            # week_date_info_tuple = tuple(df_loc.date)
            # last_index = len(week_date_info_tuple)
            # print("self.cal_day:", str(tuple(df_loc.date)[last_index - 1]))
            # # 判断长度,取最后一日
            # return str(tuple(df_loc.date)[last_index - 1])
            # 取周第一天、月的第一天
            print("self.cal_day:", str(list(df_loc.date)[0]))
            return str(list(df_loc.date)[0])

    # 1.获取原始数据
    def read_data(self):
        # 获取ods_seller相关原始表
        print("获取 ods_seller_account_syn")
        sql = f"select id as account_id, account_name, {self.week} as week from ods_seller_account_syn " \
              f"where site_name='{self.site_name}' "
        self.df_seller_acount_syn = self.spark.sql(sqlQuery=sql)
        print(sql)
        # print("self.df_seller_acount_syn:", self.df_seller_acount_syn.show(10, truncate=False))

        print("获取 ods_seller_asin_account")
        sql = f"select account_name, asin from ods_seller_asin_account " \
              f"where site_name = '{self.site_name}' group by account_name, asin"
        self.df_seller_asin_acount = self.spark.sql(sqlQuery=sql)
        print(sql)
        # print("self.df_seller_asin_acount:", self.df_seller_asin_acount.show(10, truncate=False))

        print("获取 ods_seller_account_feedback_report")
        sql = f"select account_id,num as asin_counts from ods_seller_account_feedback_report  " \
              f"where site_name='{self.site_name}' and date_type='month' and date_info='{self.year}-{self.month}'"
        self.df_seller_account_feedback_report = self.spark.sql(sqlQuery=sql)
        print(sql)
        # print("self.df_seller_account_feedback_report", self.df_seller_account_feedback_report.show(10, truncate=False))

        # 获取ods_asin相关原始表
        print("获取 ods_asin_detail_product")
        sql = f"select account_id, asin, price, rating, total_comments, {self.week} as week, row_num, created_at " \
              f"from ods_asin_detail_product where site_name='{self.site_name}' and date_type='month' and date_info='{self.year}-{self.month}'"
        self.df_asin_detail_product = self.spark.sql(sqlQuery=sql)
        print(sql)
        # print("self.df_asin_detail_product1:", self.df_asin_detail_product.show(10, truncate=False))
        self.df_asin_detail_product = self.df_asin_detail_product.sort(['account_id', "row_num", "created_at"],
                                                                       ascending=[True, True, False])
        self.df_asin_detail_product = self.df_asin_detail_product.dropDuplicates(['account_id', "row_num"])
        #print("self.df_asin_detail_product2:", self.df_asin_detail_product.show(10, truncate=False))

        print("获取 dim_asin_history_info")
        sql = f"select asin, asin_title,asin_launch_time as launch_time from dim_cal_asin_history_detail " \
              f"where site_name='{self.site_name}'"
        self.df_asin_detail = self.spark.sql(sqlQuery=sql)
        print(sql)
        # print("self.df_asin_detail:", self.df_asin_detail.show(10, truncate=False))

        print("获取 dim_asin_variation_info")
        sql = f"select asin,parent_asin from dim_asin_variation_info " \
              f"where site_name='{self.site_name}'" \
              f" and asin != parent_asin "
        self.df_asin_parent = self.spark.sql(sqlQuery=sql)
        print(sql)
        # print("self.df_asin_parent:", self.df_asin_parent.show(10, truncate=False))

    def handle_data(self):
        self.handle_asin_top20_avg()
        self.handle_asin_count()
        self.handle_save_date()

    # 2.1处理top20产品的平均指标值-按account聚合统计
    def handle_asin_top20_avg(self):
        print("处理asin_detail_product的top20指标逻辑")
        self.df_seller_top = self.df_asin_detail_product.filter("row_num<=20")
        self.df_seller_top = self.df_seller_top.groupby('account_id').agg(
            F.avg('price').alias('top_20_avg_price'),
            F.avg('rating').alias('top_20_avg_rating'),
            F.avg('total_comments').alias('top_20_avg_total_comments'),
        )
        self.df_seller_top = self.df_seller_top.withColumn("top_20_avg_total_comments",
                                                           ceil(self.df_seller_top.top_20_avg_total_comments))
        # print("df_seller_top:", self.df_seller_top.show(10, truncate=False))

    # 2.2 计算asin_count和asin_new_count逻辑
    def handle_asin_count(self):
        print("处理df_seller_account相关数据逻辑")
        # 让 df_seller_acount_syn 与 df_seller_asin_acount 和 df_seller_account_feedback_report 关联得到具体明细
        self.df_seller = self.df_seller_acount_syn. \
            join(self.df_seller_asin_acount, on='account_name', how='inner'). \
            join(self.df_asin_detail, on='asin', how='left')

        # 标记是否新品标签
        self.df_seller = self.df_seller.withColumn("days_diff", self.u_launch_time(
            self.df_seller.launch_time, F.lit(self.week_date)))

        # 通过days_diff走自定义udf,生成is_asin_new字段(是否asin新品标记)
        self.df_seller = self.df_seller.withColumn("is_asin_new", self.u_days_diff(
            self.df_seller.days_diff))

        # 做缓存
        self.df_seller = self.df_seller.cache()

        # 计算店铺-asin的打包数量
        self.df_seller = self.df_seller.withColumn('asin_package_quantity', self.u_judge_package_quantity(F.col('asin_title')))
        # 符合打包数量>=2的商品数标识
        self.df_seller = self.df_seller.withColumn('is_pq_flag', F.when(F.col('asin_package_quantity') >= 2, F.lit(1)))

        # 计算asin_counts_exists与asin_new_counts指标
        self.df_asin_counts = self.df_seller.groupby(['account_id', 'account_name']).agg(
            F.count('asin').alias('asin_counts_exists'),
            F.sum('is_asin_new').alias('asin_new_counts'),
            F.sum('is_pq_flag').alias('fb_package_quantity_num')
        )

        # 关联report表,获取店铺的商品数量
        self.df_asin_counts = self.df_asin_counts\
            .join(self.df_seller_account_feedback_report, on='account_id', how='left')

        # 计算 counts_new_rate asin新品比率
        self.df_asin_counts = self.df_asin_counts.withColumn('counts_new_rate',
                                                             F.round(F.col('asin_new_counts')/F.col('asin_counts_exists'), 3))

        # 计算店铺打包数量的比例
        self.df_asin_counts = self.df_asin_counts.withColumn('fb_package_quantity_prop',
                                                             F.round(F.col('fb_package_quantity_num') / F.col('asin_counts_exists'),
                                                                     3))

        # 计算店铺多数量占比 有变体asin数量/(有变体asin数量+单产品asin数量) 逻辑实现
        self.df_variat_ratio = self.df_seller.join(self.df_asin_parent, on='asin', how='left')

        # 没有匹配上asin_parent的给他附上自己的asin值:1.可能是父asin最顶端; 2.可能是asin单产品
        self.df_variat_ratio = self.df_variat_ratio.withColumn('parent_asin_new',
                                                               F.when(F.col('parent_asin').isNull(), F.col('asin'))
                                                               .otherwise(F.col('parent_asin')))

        # 按照account_name、asin_parent分组统计数量
        self.df_variat_ratio = self.df_variat_ratio.groupby(['account_id', 'account_name', 'parent_asin_new'])\
            .agg(
            F.count('asin').alias('asin_son_count')
        )

        # 打上多变体标签
        self.df_variat_ratio = self.df_variat_ratio.withColumn('is_variat_flag',
                                                               F.when(F.col('asin_son_count') > 1, F.lit(1)))

        # 按照account_name分组,得出分子分母
        self.df_variat_ratio = self.df_variat_ratio.groupby(['account_id', 'account_name'])\
            .agg(
            F.sum('is_variat_flag').alias('variat_num'),
            F.count('parent_asin_new').alias('total_asin_num'))

        self.df_variat_ratio = self.df_variat_ratio.na.fill({'variat_num': 0, 'total_asin_num': 0})
        ## 计算店铺多数量占比 有变体asin数量/(有变体asin数量+单产品asin数量)
        self.df_variat_ratio = self.df_variat_ratio.withColumn('fb_variat_prop',
                                                               F.round(F.col('variat_num')/F.col('total_asin_num'), 3))

        self.df_variat_ratio = self.df_variat_ratio.drop('account_name')

    # 2.4 指标整合逻辑
    def handle_save_date(self):
        # seller_detail
        self.df_seller = self.df_seller\
            .select(
            F.col('account_id'),
            F.col('account_name'),
            F.col('asin'),
            F.col('launch_time'),
            F.col('days_diff'),
            F.col('is_asin_new'),
            # 遗留的无用字段
            F.lit(0).alias('asin_counts'),
            F.lit(0).alias('asin_new_counts'),
            F.lit(0).alias('counts_new_rate'),
            F.lit(0).alias('asin_counts_exists'),

            # 分区字段补全
            F.lit(self.week).cast('int').alias('week'),
            F.lit(self.site_name).alias("site_name"),
            F.lit(self.date_type).alias("date_type"),
            F.lit(self.date_info).alias("date_info")
                    )

        # 关联top20avg相关计算指标以及计算店铺多数量占比计算指标
        self.df_seller_agg = self.df_asin_counts\
            .join(self.df_seller_top, on='account_id', how='left')\
            .join(self.df_variat_ratio, on='account_id', how='left')

        self.df_seller_agg = self.df_seller_agg.select(
            F.col('account_id'),
            F.col('account_name'),
            F.col('asin_new_counts'),
            F.col('asin_counts'),
            F.col('counts_new_rate'),
            F.col('top_20_avg_price'),
            F.col('top_20_avg_rating'),
            F.col('top_20_avg_total_comments'),
            F.col('asin_counts_exists'),
            F.col('variat_num').alias('fb_variat_num'),
            F.col('total_asin_num').alias('fb_asin_total'),
            F.col('fb_variat_prop'),
            F.col('fb_package_quantity_prop'),
            F.col('fb_package_quantity_num'),
            # 分区字段补全
            F.lit(self.week).cast('int').alias('week'),
            F.lit(self.ym).alias('ym'),
            F.lit(self.site_name).alias("site_name"),
            F.lit(self.date_type).alias("date_type"),
            F.lit(self.date_info).alias("date_info")
        )


    # 重写数据写入方法,对应两张目标表
    def save_data(self):
        self.reset_partitions(partitions_num=1)
        self.save_data_common(
            df_save=self.df_seller_agg,
            db_save='dwd_seller_asin_account_agg',
            partitions_num=self.partitions_num,
            partitions_by=self.partitions_by
        )

        self.reset_partitions(partitions_num=10)
        self.save_data_common(
            df_save=self.df_seller,
            db_save='dwd_seller_asin_account_detail',
            partitions_num=self.partitions_num,
            partitions_by=self.partitions_by
        )


if __name__ == '__main__':
    site_name = sys.argv[1]  # 参数1:站点
    date_type = sys.argv[2]  # 参数2:类型:week/4_week/month/quarter
    date_info = sys.argv[3]  # 参数3:年-周/年-月/年-季, 比如: 2022-1
    handle_obj = DwdFeedBack(site_name=site_name, date_type=date_type, date_info=date_info)
    handle_obj.run()