"""
@Author : wangrui
@Description : 流量选品
@SourceTable :
dwd_asin_measure
dim_asin_detail
ods_bsr_end
dim_asin_bs_category
dim_fd_asin_info
dim_asin_volume
@SinkTable : dwt_flow_asin
@CreateTime : 2023/01/10 07:55
@UpdateTime : 2023/01/10 07:55
"""
import os
import sys
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from utils.templates import Templates
# 分组排序的udf窗口函数
from pyspark.sql.window import Window
from datetime import datetime, timedelta
from pyspark.sql import functions as F
from utils.db_util import DBUtil
from utils.spark_util import SparkUtil
from utils.common_util import CommonUtil
from utils.hdfs_utils import HdfsUtils
from pyspark.storagelevel import StorageLevel
from pyspark.sql.types import *
from utils.DorisHelper import DorisHelper
class DwtFlowAsin(Templates):
def __init__(self, site_name="us", date_type="week", date_info="2022-1"):
super().__init__()
self.site_name = site_name
self.date_type = date_type
self.date_info = date_info
self.hive_tb = f"dwt_flow_asin"
self.partition_dict = {
"site_name": site_name,
"date_type": date_type,
"date_info": date_info
}
self.hdfs_path = CommonUtil.build_hdfs_path(self.hive_tb, partition_dict=self.partition_dict)
self.spark = self.create_spark_object(
app_name=f"{self.hive_tb}: {self.site_name}, {self.date_type}, {self.date_info}")
self.previous_date = self.udf_get_previous_last_30_day(self)
self.current_date = self.udf_get_current_time(self)
# doris相关配置
self.doris_db = "test" if self.test_flag == "test" else "selection"
self.asin_latest_detail_table = f"{self.site_name}_asin_latest_detail"
# 写入、分区初始化
self.df_save = self.spark.sql(f"select 1+1;")
self.partitions_by = ['site_name', 'date_type', 'date_info']
self.reset_partitions(60)
self.launch_time_interval_dict = self.get_launch_time_interval_dict()
# 初始化全局df
self.df_asin_detail = self.spark.sql(f"select 1+1;")
self.df_asin_measure = self.spark.sql(f"select 1+1;")
self.df_bsr_end = self.spark.sql(f"select 1+1;")
self.df_asin_bs_category = self.spark.sql(f"select 1+1;")
self.df_fd_asin_info = self.spark.sql(f"select 1+1;")
self.df_flow_asin_last = self.spark.sql(f"select 1+1;")
self.df_title_matching_degree = self.spark.sql(f"select 1+1;")
@staticmethod
def udf_get_previous_last_30_day(self):
self.df_date = self.spark.sql(f"select * from dim_date_20_to_30 ;")
df = self.df_date.toPandas()
if self.date_type == 'last30day':
df_loc = df.loc[df.date == f'{self.date_info}']
current_date_id = list(df_loc.id)[0]
original_date_id = int(current_date_id) - 30
df_loc = df.loc[df.id == original_date_id]
original_year_month = list(df_loc.year_month)[0]
df_loc = df.loc[(df.year_month == original_year_month) & (df.day == 1)]
original_year_month_id = list(df_loc.id)[0]
previous_year_month_id = int(original_year_month_id) - 1
df_loc = df.loc[df.id == previous_year_month_id]
previous_date = list(df_loc.year_month)[0]
return previous_date
elif self.date_type in ['month', 'month_week']:
df_loc = df.loc[(df.year_month == f'{self.date_info}') & (df.day == 1)]
current_month_id = list(df_loc.id)[0]
previous_month_id = int(current_month_id) - 1
df_loc = df.loc[df.id == previous_month_id]
previous_date = list(df_loc.year_month)[0]
return previous_date
elif self.date_type == '4_week':
df_loc = df.loc[(df.year_week == f'{self.date_info}') & (df.week_day == 1)]
current_4_week_id = list(df_loc.id)[0]
df_loc = df.loc[df.id == int(current_4_week_id) - 21]
current_4_week_month = list(df_loc.year_month)[0]
df_loc = df.loc[(df.year_month == current_4_week_month) & (df.day == 1)]
current_4_week_month_id = list(df_loc.id)[0]
previous_4_week_month_id = int(current_4_week_month_id) - 1
df_loc = df.loc[df.id == previous_4_week_month_id]
previous_date = list(df_loc.year_month)[0]
return previous_date
@staticmethod
def udf_get_current_time(self):
self.df_date = self.spark.sql(f"select * from dim_date_20_to_30 ;")
df = self.df_date.toPandas()
if self.date_type == 'month':
df_loc = df.loc[(df.year_month == f'{self.date_info}') & (df.day == 25)]
current_id = list(df_loc.id)[0]
df_loc = df.loc[df.id == current_id]
current_date = list(df_loc.date)[0]
return current_date
elif self.date_type in ['week', '4_week']:
df_loc = df.loc[(df.year_week == f'{self.date_info}') & (df.week_day == 7)]
current_id = list(df_loc.id)[0]
df_loc = df.loc[df.id == current_id]
current_date = list(df_loc.date)[0]
return current_date
elif self.date_type in ['last30day', 'day']:
return self.date_info
elif self.date_type in ['month_week']:
return str(datetime.now().date())
@staticmethod
def get_launch_time_interval_dict():
cur_date = datetime.now().date()
return {
"one_month": (cur_date + timedelta(days=-30)).strftime('%Y-%m-%d'),
"three_month": (cur_date + timedelta(days=-90)).strftime('%Y-%m-%d'),
"six_month": (cur_date + timedelta(days=-180)).strftime('%Y-%m-%d'),
"twelve_month": (cur_date + timedelta(days=-360)).strftime('%Y-%m-%d'),
"twenty_four_month": (cur_date + timedelta(days=-720)).strftime('%Y-%m-%d'),
"thirty_six_month": (cur_date + timedelta(days=-1080)).strftime('%Y-%m-%d')
}
@staticmethod
def calculate_change(current_col, previous_col):
rise_col = F.col(current_col) - F.col(previous_col)
change_col = F.when((F.col(previous_col).isNotNull()) & (F.col(previous_col) != 0),
F.round((F.col(current_col) - F.col(previous_col)) / F.col(previous_col), 4)
).otherwise(None)
return rise_col, change_col
def read_data(self):
print("1.获取dwd_asin_measure,得到各种类型的统计、ao值、自然流量占比、月销信息、bsr销量信息")
sql = f"""
select asin, round(asin_ao_val, 3) as asin_ao_val, round(asin_ao_val_matrix, 3) as matrix_ao_val,
asin_zr_counts, asin_sp_counts, (asin_sb1_counts + asin_sb2_counts) as asin_sb_counts,
asin_sb3_counts as asin_vi_counts, asin_bs_counts, asin_ac_counts, asin_tr_counts, asin_er_counts, asin_st_counts,
asin_bsr_orders as bsr_orders, asin_amazon_orders, round(asin_zr_flow_proportion, 3) as zr_flow_proportion,
round(asin_flow_proportion_matrix, 3) as matrix_flow_proportion
from dwd_asin_measure where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}'"""
print("sql:" + sql)
self.df_asin_measure = self.spark.sql(sqlQuery=sql)
self.df_asin_measure = self.df_asin_measure.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_measure.show(10, truncate=False)
print("2.获取dim_asin_detail,得到asin详情")
sql = f"""
select asin, asin_img_url, lower(asin_title) as asin_title, asin_title_len, asin_price, asin_rating, asin_total_comments,
asin_buy_box_seller_type, seller_json, asin_page_inventory, asin_category_desc, asin_volume, asin_weight, asin_length, asin_width, asin_height,
asin_color, asin_size, asin_style, asin_is_sale, asin_launch_time, asin_is_new, asin_img_num, asin_img_type, asin_material,
lower(asin_brand_name) as asin_brand_name, asin_activity_type, act_one_two_val, act_three_four_val, act_five_six_val, act_eight_val,
one_star,two_star, three_star, four_star, five_star, low_star, together_asin, ac_name, variation_num, account_name, account_id, parent_asin,
asin_lob_info, is_contains_lob_info, is_package_quantity_abnormal, asin_quantity_variation_type, package_quantity,
asin_is_movie as is_movie_label, asin_is_brand as is_brand_label, asin_is_alarm as is_alarm_brand, asin_is_self,
date_format(created_time, 'yyyy-MM-dd HH:mm:ss') as asin_crawl_date, asin_bought_month, asin_image_view,
case when product_description is not null then 1 else 0 end as is_with_product_description, asin_describe,
category_id as top_category_id, category_first_id as top_category_first_id, customer_reviews_json, img_list as img_info,
asin_follow_sellers as follow_sellers_count
from dim_asin_detail where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"""
print("sql:" + sql)
self.df_asin_detail = self.spark.sql(sqlQuery=sql)
self.df_asin_detail = self.df_asin_detail.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_detail.show(10, truncate=False)
print("3.获取ods_bsr_end,获取有效rank信息")
sql = f"""select rank as limit_rank, category_id as category_first_id from ods_bsr_end where site_name='{self.site_name}'"""
print("sql:" + sql)
df_bsr_end = self.spark.sql(sqlQuery=sql)
self.df_bsr_end = F.broadcast(df_bsr_end)
self.df_bsr_end.show(10, truncate=False)
print("4.获取dim_asin_bs_category,获取分类名称")
sql = f"""
select asin, asin_bs_cate_1_rank as first_category_rank, asin_bs_cate_current_rank as current_category_rank,
asin_bs_cate_1_id as category_first_id, asin_bs_cate_current_id as category_id
from dim_asin_bs_info where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info = '{self.date_info}'"""
print("sql:" + sql)
self.df_asin_bs_category = self.spark.sql(sqlQuery=sql)
self.df_asin_bs_category = self.df_asin_bs_category.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_bs_category.show(10, truncate=False)
print("5.获取dim_fd_asin_info,得到卖家相关信息")
if (self.date_type in ['month', 'month_week'] and self.date_info >= '2024-05') or (self.date_type == '4_week' and self.date_info >= '2024-21'):
sql = f"""
select fd_unique as account_id, upper(fd_country_name) as seller_country_name from dim_fd_asin_info
where site_name='{self.site_name}' and fd_unique is not null group by fd_unique, fd_country_name"""
else:
sql = f"""
select account_id, account_name, seller_country_name, asin
from (select fd_unique as account_id, fd_account_name as account_name, upper(fd_country_name) as seller_country_name, asin,
ROW_NUMBER() OVER (PARTITION BY asin ORDER BY updated_at DESC) AS t_rank
from dim_fd_asin_info where site_name = '{self.site_name}' and fd_unique is not null) tmp
where tmp.t_rank = 1
"""
self.df_fd_asin_info = self.spark.sql(sqlQuery=sql)
self.df_fd_asin_info = self.df_fd_asin_info.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_fd_asin_info.show(10, truncate=False)
print("6.获取上一个最近30天的整合结果")
sql = f"""
select asin, round(asin_ao_val, 3) as previous_asin_ao_val, asin_price as previous_asin_price,
sales as pervious_sales, variation_num as previous_variation_num, asin_rating as previous_asin_rating,
bsr_orders as previous_bsr_orders, asin_total_comments as previous_asin_total_comments,
first_category_rank as previous_first_category_rank from dwt_flow_asin
where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.previous_date}'
"""
print("sql:" + sql)
self.df_flow_asin_last = self.spark.sql(sqlQuery=sql)
self.df_flow_asin_last = self.df_flow_asin_last.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_flow_asin_last.show(10, truncate=False)
print("7.获取asin的标题匹配度")
sql = f"""
select asin, contains_flag from dwd_title_matching_degree where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}'
"""
print("sql:" + sql)
self.df_title_matching_degree = self.spark.sql(sqlQuery=sql)
self.df_title_matching_degree = self.df_title_matching_degree.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_title_matching_degree.show(10, truncate=False)
# 处理asin基础属性信息(体积重量相关)及bsr销售额相关信息
def handle_asin_basic_attribute(self):
# 处理重量类型
self.df_asin_detail = self.df_asin_detail.withColumn("asin_weight_type", F.expr("""CASE
WHEN asin_weight BETWEEN 0 AND 0.2 THEN 1
WHEN asin_weight BETWEEN 0.2 AND 0.4 THEN 2
WHEN asin_weight BETWEEN 0.4 AND 0.6 THEN 3
WHEN asin_weight BETWEEN 0.6 AND 1 THEN 4
WHEN asin_weight BETWEEN 1 AND 2 THEN 5
WHEN asin_weight >= 2 THEN 6
ELSE 0 END"""))
# 处理体积重/毛重相关信息
self.df_asin_detail = self.df_asin_detail.withColumn(
"asin_weight_ratio",
F.when(
F.col("asin_length").isNotNull() & (F.col("asin_width").isNotNull()) & (F.col("asin_height").isNotNull()) & (F.col("asin_weight") > 0),
F.round(F.col("asin_length") * F.col("asin_width") * F.col("asin_height") * 3.2774128 / (F.col("asin_weight") * 453.59), 4)
).otherwise(F.lit(-1))
)
# 处理尺寸类型
if self.site_name == 'us':
expr_str = f"""
CASE WHEN asin_weight > 0 AND asin_weight * 16 <= 16 AND asin_length > 0 AND asin_length <= 15 AND asin_width > 0 AND asin_width <= 12 AND asin_height > 0 AND asin_height <= 0.75 THEN 1
WHEN asin_weight > 0 AND asin_weight <= 20 AND asin_length > 0 AND asin_length <= 18 AND asin_width > 0 AND asin_width <= 14 AND asin_height > 0 AND asin_height <= 8 THEN 2
WHEN asin_weight > 0 AND asin_weight <= 70 AND asin_length > 0 AND asin_length <= 60 AND asin_width > 0 AND asin_width <= 30 AND asin_length + asin_length + (asin_width + asin_height) * 2 <= 130 THEN 3
WHEN asin_weight > 0 AND asin_weight <= 150 AND asin_length > 0 AND asin_length <= 108 AND asin_length + asin_length + (asin_width + asin_height) * 2 <= 130 THEN 4
WHEN asin_weight > 0 AND asin_weight <= 150 AND asin_length > 0 AND asin_length <= 108 AND asin_length + asin_length + (asin_width + asin_height) * 2 <= 165 THEN 5
WHEN asin_weight > 150 AND asin_length > 108 AND asin_length + asin_length + (asin_width + asin_height) * 2 > 165 THEN 6 ELSE 0 END"""
else:
expr_str = f"""
CASE
WHEN asin_weight > 0 AND asin_weight <= 100 AND asin_length > 0 AND asin_length <= 20 AND asin_width > 0 AND asin_width <= 15 AND asin_height > 0 AND asin_height <= 1 THEN 1
WHEN asin_weight > 0 AND asin_weight <= 500 AND asin_length > 0 AND asin_length <= 33 AND asin_width > 0 AND asin_width <= 23 AND asin_height > 0 AND asin_height <= 2.5 THEN 2
WHEN asin_weight > 0 AND asin_weight <= 1000 AND asin_length > 0 AND asin_length <= 33 AND asin_width > 0 AND asin_width <= 23 AND asin_height > 0 AND asin_height <= 5 THEN 3
WHEN asin_weight > 0 AND asin_weight <= 12000 AND asin_length > 0 AND asin_length <= 45 AND asin_width > 0 AND asin_width <= 34 AND asin_height > 0 AND asin_height <= 26 THEN 4
WHEN asin_weight > 0 AND asin_weight <= 2000 AND asin_length > 0 AND asin_length <= 61 AND asin_width > 0 AND asin_width <= 46 AND asin_height > 0 AND asin_height <= 46 THEN 5
WHEN asin_length > 0 AND asin_length <= 150 AND asin_length + asin_length + (asin_width + asin_height) <= 300 THEN 6
WHEN asin_length > 150 AND asin_length + asin_length + (asin_width + asin_height) > 300 THEN 7
ELSE 0 END"""
self.df_asin_detail = self.df_asin_detail.withColumn("asin_size_type", F.expr(expr_str)).drop("asin_length", "asin_width", "asin_height")
# 通过ASIN页面信息处理(评分类型、上架时间类型、价格类型)
def handle_asin_detail_all_type(self):
# 评分类型
self.df_asin_detail = self.df_asin_detail.withColumn(
"asin_rating_type", F.expr("""CASE WHEN asin_rating >= 4.5 THEN 1 WHEN asin_rating >= 4 AND asin_rating < 4.5 THEN 2 WHEN asin_rating >= 3.5 AND asin_rating < 4 THEN 3
WHEN asin_rating >= 3 AND asin_rating < 3.5 THEN 4 WHEN asin_rating < 3 AND asin_rating >= 0 THEN 5 ELSE 0 END"""))
# 上架时间类型
one_month = self.launch_time_interval_dict['one_month']
three_month = self.launch_time_interval_dict['three_month']
six_month = self.launch_time_interval_dict['six_month']
twelve_month = self.launch_time_interval_dict['twelve_month']
twenty_four_month = self.launch_time_interval_dict['twenty_four_month']
thirty_six_month = self.launch_time_interval_dict['thirty_six_month']
expr_str = f"""CASE WHEN asin_launch_time >= '{one_month}' THEN 1
WHEN asin_launch_time >= '{three_month}' AND asin_launch_time < '{one_month}' THEN 2
WHEN asin_launch_time >= '{six_month}' AND asin_launch_time < '{three_month}' THEN 3
WHEN asin_launch_time >= '{twelve_month}' AND asin_launch_time < '{six_month}' THEN 4
WHEN asin_launch_time >= '{twenty_four_month}' AND asin_launch_time < '{twelve_month}' THEN 5
WHEN asin_launch_time >= '{thirty_six_month}' AND asin_launch_time < '{twenty_four_month}' THEN 6
WHEN asin_launch_time < '{thirty_six_month}' THEN 7 ELSE 0 END"""
self.df_asin_detail = self.df_asin_detail.withColumn("asin_launch_time_type", F.expr(expr_str))
# 价格类型
self.df_asin_detail = self.df_asin_detail.withColumn(
"asin_price_type", F.expr("""
CASE WHEN asin_price IS NOT NULL AND asin_price > 0 AND asin_price < 10 THEN 1
WHEN asin_price >= 10 AND asin_price < 15 THEN 2
WHEN asin_price >= 15 AND asin_price < 20 THEN 3
WHEN asin_price >= 20 AND asin_price < 30 THEN 4
WHEN asin_price >= 30 AND asin_price < 50 THEN 5
WHEN asin_price >= 50 THEN 6 ELSE 0 END"""))
# 处理asin分类、排名、排名类型字段、是否有效排名信息
def handle_asin_category_info(self):
self.df_asin_detail = self.df_asin_detail.join(self.df_asin_bs_category, on=['asin'], how='left')
self.df_asin_detail = self.df_asin_detail.withColumn(
"category_id", F.coalesce(F.col("category_id"), F.col("top_category_id"))
).withColumn(
"category_first_id", F.coalesce(F.col("category_first_id"), F.col("top_category_first_id"))
).drop("top_category_id", "top_categoty_first_id")
self.df_asin_detail = self.df_asin_detail.withColumn(
"asin_rank_type", F.expr("""
CASE WHEN first_category_rank IS NOT NULL AND first_category_rank BETWEEN 0 AND 1000 THEN 1
WHEN first_category_rank BETWEEN 1000 AND 5000 THEN 2
WHEN first_category_rank BETWEEN 5000 AND 10000 THEN 3
WHEN first_category_rank BETWEEN 10000 AND 20000 THEN 4
WHEN first_category_rank BETWEEN 20000 AND 30000 THEN 5
WHEN first_category_rank BETWEEN 30000 AND 50000 THEN 6
WHEN first_category_rank BETWEEN 50000 AND 70000 THEN 7
WHEN first_category_rank >= 70000 THEN 8 ELSE 0 END"""))
self.df_asin_detail = self.df_asin_detail.join(self.df_bsr_end, on=['category_first_id'], how='left')
self.df_asin_detail = self.df_asin_detail.withColumn(
"bsr_type", F.expr("""CASE WHEN limit_rank is null and category_first_id <= 500000 THEN 1 WHEN limit_rank is not null and category_first_id <= limit_rank THEN 1 ELSE 0 END""")).drop("limit_rank")
self.df_asin_bs_category.unpersist()
# 处理asin的bsr销量、亚马逊月销信息、ao值、母体ao、自然流量占比、母体自然流量占比、ao值类型
def handle_asin_measure(self):
self.df_asin_detail = self.df_asin_detail.join(self.df_asin_measure, on=['asin'], how='left')
self.df_asin_detail = self.df_asin_detail.withColumn(
"sales", F.round(F.col("bsr_orders") * F.col("asin_price"), 2)
).withColumn(
"asin_bought_month", F.coalesce(F.col("asin_bought_month"), F.col("asin_amazon_orders"))
).withColumn("asin_ao_val_type", F.expr("""CASE WHEN asin_ao_val BETWEEN 0 AND 0.1 THEN 1
WHEN asin_ao_val BETWEEN 0.1 AND 0.2 THEN 2 WHEN asin_ao_val BETWEEN 0.2 AND 0.4 THEN 3
WHEN asin_ao_val BETWEEN 0.4 AND 0.8 THEN 4 WHEN asin_ao_val BETWEEN 0.8 AND 1.2 THEN 5
WHEN asin_ao_val BETWEEN 1.2 AND 2 THEN 6 WHEN asin_ao_val >= 2 THEN 7 ELSE 0 END""")).drop("asin_amazon_orders")
self.df_asin_measure.unpersist()
# 处理配送方式、卖家所在地以及卖家所在地类型
def handle_seller_country(self):
if (self.date_type in ['month', 'month_week'] and self.date_info >= '2024-05') or (self.date_type == '4_week' and self.date_info >= '2024-21'):
self.df_asin_detail = self.df_asin_detail.join(self.df_fd_asin_info, on=['account_id'], how='left')
else:
self.df_asin_detail = self.df_asin_detail.drop("account_id", "account_name")
self.df_asin_detail = self.df_asin_detail.join(self.df_fd_asin_info, on=['asin'], how='left')
self.df_asin_detail = self.df_asin_detail.withColumn("asin_site_name_type", F.expr("""
CASE WHEN asin_buy_box_seller_type = 1 THEN 4
WHEN asin_buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%US%' THEN 1
WHEN asin_buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%CN%' THEN 2 ELSE 3 END"""))
self.df_fd_asin_info.unpersist()
# 处理asin的lqs评分
def handle_asin_lqs_rating(self):
self.df_asin_detail = self.df_asin_detail.withColumn(
"category_node_rating", F.when(F.col("category_id").isNotNull(), F.lit(1)).otherwise(F.lit(0))
).withColumn(
"zr_rating", F.when(F.col("asin_zr_counts") > 0, F.lit(0.5)).otherwise(F.lit(0))
).withColumn(
"sp_rating", F.when(F.col("asin_sp_counts") > 0, F.lit(1)).otherwise(F.lit(0))
).withColumn(
"a_add_rating", F.when(F.col("asin_img_type").contains("3"), F.lit(1)).otherwise(F.lit(0))
).withColumn(
"video_rating", F.when(F.col("asin_img_type").contains("2"), F.lit(0.5)).otherwise(F.lit(0))
).withColumn(
"brand_rating", F.when(F.col("is_brand_label") == 1, F.lit(0.2)).otherwise(F.lit(0))
).withColumn(
"product_describe_rating", F.when(F.col("is_with_product_description") == 1, F.lit(0.2)).otherwise(F.lit(0))
).withColumn(
"highlight_rating",
F.when((F.col("asin_describe").isNotNull()) & (F.size(F.split(F.col("asin_describe"), '\\|-\\|')) <= 4),
F.size(F.split(F.col("asin_describe"), '\\|-\\|')) * 0.4)
.when((F.col("asin_describe").isNotNull()) & (F.size(F.split(F.col("asin_describe"), '\\|-\\|')) > 4),
F.lit(1.6))
.otherwise(F.lit(0))
).withColumn(
"title_len_rating",
F.when((F.col("asin_title_len") >= 50) & (F.col("asin_title_len") <= 200), F.lit(0.5)).otherwise(F.lit(0))
).withColumn(
"title_brand_rating",
F.expr(f"""CASE WHEN asin_brand_name is not null AND lower(regexp_replace(asin_title, '[^a-zA-Z0-9\\s]', '')) LIKE lower(regexp_replace(asin_brand_name, '[^a-zA-Z0-9\\s]', '')) || '%' THEN 0.5 ELSE 0 END""")
).withColumn(
"img_num_rating", F.when(F.col("asin_img_num") <= 4, F.col("asin_img_num") * 0.5).when(F.col("asin_img_num") > 4, F.lit(2)).otherwise(F.lit(0))
).withColumn(
"img_enlarge_rating", F.when(F.col("asin_image_view") == 1, F.lit(0.5)).otherwise(F.lit(0))
)
self.df_asin_detail = self.df_asin_detail.withColumn(
"asin_lqs_rating_detail",
F.to_json(F.struct(F.col("category_node_rating"), F.col("zr_rating"), F.col("sp_rating"),
F.col("a_add_rating"), F.col("video_rating"), F.col("brand_rating"),
F.col("product_describe_rating"), F.col("highlight_rating"), F.col("title_len_rating"),
F.col("title_brand_rating"), F.col("img_num_rating"), F.col("img_enlarge_rating"))
)
)
self.df_asin_detail = self.df_asin_detail.withColumn(
"asin_lqs_rating",
F.col("category_node_rating") + F.col("zr_rating") + F.col("sp_rating") + F.col("a_add_rating") + F.col(
"video_rating") + F.col("brand_rating") + F.col("product_describe_rating") + F.col(
"highlight_rating") + F.col("title_len_rating") + F.col("title_brand_rating") + F.col(
"img_num_rating") + F.col("img_enlarge_rating")
)
self.df_asin_detail = self.df_asin_detail.\
drop("is_with_product_description", "asin_describe", "asin_image_view", "category_node_rating", "zr_rating",
"sp_rating", "a_add_rating", "video_rating", "brand_rating", "product_describe_rating",
"highlight_rating", "title_len_rating", "title_brand_rating", "img_num_rating", "img_enlarge_rating")
# 处理asin是否隐藏分类信息（US/UK/DE站点通用）以及asin_type信息
def handle_asin_is_hide(self):
mysql_con = DBUtil.get_connection_info("mysql", "us")
sql = f"""
select category_id_base as category_id, 1 as hide_flag from us_bs_category_hide group by category_id_base
"""
df_hide_category = SparkUtil.read_jdbc_query(session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
username=mysql_con['username'], query=sql)
self.df_asin_detail = self.df_asin_detail.join(df_hide_category, on=['category_id'], how='left')
self.df_asin_detail = self.df_asin_detail.withColumn("asin_is_hide", F.expr("""
CASE WHEN hide_flag = 1 THEN 1 WHEN category_first_id = 'grocery' and category_id != '6492272011' THEN 1
WHEN category_id in ('21393128011', '21377129011', '21377127011', '21377130011', '21388218011', '21377132011') THEN 1
ELSE 0 END""")).drop("hide_flag")
self.df_asin_detail = self.df_asin_detail.withColumn("asin_is_need", F.expr("""
CASE WHEN category_first_id in ('mobile-apps', 'audible', 'books', 'music', 'dmusic', 'digital-text', 'magazines', 'movies-tv', 'software', 'videogames', 'amazon-devices', 'boost', 'us-live-explorations', 'amazon-renewed') THEN 1
WHEN asin NOT LIKE 'B0%' THEN 1
ELSE 0 END"""))
self.df_asin_detail = self.df_asin_detail.withColumn("asin_type", F.expr("""
CASE WHEN asin_is_self=1 THEN 1 WHEN asin_is_need=1 THEN 2 WHEN asin_is_hide=1 THEN 3 ELSE 0 END"""
)).drop("asin_is_self", "asin_is_need", "asin_is_hide")
# 处理匹配度
def handle_title_matching_degree(self):
window = Window.partitionBy("asin")
self.df_title_matching_degree = self.df_title_matching_degree.withColumn(
"asin_count", F.count("asin").over(window))
# 统计每个asin中，flag为1的数量
self.df_title_matching_degree = self.df_title_matching_degree.withColumn(
"contains_count", F.sum("contains_flag").over(window))
self.df_title_matching_degree = self.df_title_matching_degree.withColumn(
"title_matching_degree", F.round(F.col("contains_count") / F.col("asin_count"), 4))
self.df_title_matching_degree = self.df_title_matching_degree.drop("asin_count", "contains_count", "contains_flag")
self.df_title_matching_degree = self.df_title_matching_degree.drop_duplicates(['asin'])
self.df_asin_detail = self.df_asin_detail.join(self.df_title_matching_degree, on=['asin'], how='left')
self.df_title_matching_degree.unpersist()
# 处理变体ASIN属性(asin维度)的变化率相关信息
def handle_asin_attribute_change(self):
self.df_asin_detail = self.df_asin_detail.join(self.df_flow_asin_last, on=['asin'], how='left')
columns_to_change = [
("first_category_rank", "previous_first_category_rank", "asin_rank"),
("bsr_orders", "previous_bsr_orders", "asin_bsr_orders"),
("asin_rating", "previous_asin_rating", "asin_rating"),
("asin_total_comments", "previous_asin_total_comments", "asin_comments"),
("variation_num", "previous_variation_num", "asin_variation"),
("asin_ao_val", "previous_asin_ao_val", "asin_ao"),
("asin_price", "previous_asin_price", "asin_price"),
("sales", "pervious_sales", "asin_sales")
]
for current_col, previous_col, suffix in columns_to_change:
rise_col, change_col = self.calculate_change(current_col, previous_col)
if suffix == 'asin_ao':
self.df_asin_detail = self.df_asin_detail.withColumn(f"{suffix}_rise", F.round(rise_col, 3))
elif suffix in ['asin_price', 'sales']:
self.df_asin_detail = self.df_asin_detail.withColumn(f"{suffix}_rise", F.round(rise_col, 2))
elif suffix == 'asin_rating':
self.df_asin_detail = self.df_asin_detail.withColumn(f"{suffix}_rise", F.round(rise_col, 1))
else:
self.df_asin_detail = self.df_asin_detail.withColumn(f"{suffix}_rise", rise_col.cast(IntegerType()))
self.df_asin_detail = self.df_asin_detail.withColumn(f"{suffix}_change", F.round(change_col, 4))
self.df_asin_detail.drop(previous_col)
self.df_flow_asin_last.unpersist()
# 字段标准化
def handle_column(self):
self.df_save = self.df_asin_detail.\
select("asin", "asin_ao_val", "asin_zr_counts", "asin_sp_counts", "asin_sb_counts", "asin_vi_counts",
"asin_bs_counts", "asin_ac_counts", "asin_tr_counts", "asin_er_counts", "bsr_orders",
F.lit(None).alias("orders"), "sales", F.lit(None).alias("cate_current_id"),
F.lit(None).alias("cate_1_id"), "asin_img_url", "asin_title", "asin_title_len", "asin_price",
"asin_rating", "asin_total_comments", "asin_buy_box_seller_type", "asin_page_inventory",
"asin_category_desc", "asin_volume", "asin_weight", "asin_color", "asin_size", "asin_style",
"asin_is_sale", F.lit(None).alias("asin_rank"), "asin_launch_time", "asin_is_new", "asin_img_num",
"asin_img_type", "asin_material", "asin_brand_name", "asin_activity_type", "act_one_two_val",
"act_three_four_val", "act_five_six_val", "act_eight_val", F.lit(None).alias("qa_num"), "one_star",
"two_star", "three_star", "four_star", "five_star", "low_star", "together_asin", "ac_name",
"variation_num", "account_name", "account_id", "seller_country_name", "bsr_type",
F.lit(-1).alias("bsr_best_orders_type"), F.lit(None).alias("zr_best_orders_type"), "parent_asin", "asin_rank_rise",
"asin_rank_change", "asin_ao_rise", "asin_ao_change", "asin_price_rise", "asin_price_change",
F.lit(None).alias("asin_orders_rise"), F.lit(None).alias("asin_orders_change"), "asin_rating_rise",
"asin_rating_change", "asin_comments_rise", "asin_comments_change", "asin_bsr_orders_rise",
"asin_bsr_orders_change", "asin_sales_rise", "asin_sales_change", "asin_variation_rise",
"asin_variation_change", "asin_size_type", "asin_rating_type", "asin_site_name_type",
"asin_weight_type", "asin_launch_time_type", "asin_ao_val_type", "asin_rank_type", "asin_price_type",
F.lit(None).alias("created_time"), F.lit(None).alias("updated_time"), "asin_lob_info",
"customer_reviews_json", "img_info", "is_contains_lob_info", "is_package_quantity_abnormal",
"asin_st_counts", "asin_quantity_variation_type", "package_quantity",
F.lit(None).alias("sp_type1"), F.lit(None).alias("sp_type2"), F.lit(None).alias("sp_type3"),
"is_movie_label", "is_brand_label", "is_alarm_brand", "asin_type", F.lit(None).alias("asin_cost_fee"),
F.lit(None).alias("asin_refund_fee"), F.lit(None).alias("asin_adv_fee"),
F.lit(None).alias("asin_commission_fee"), F.lit(None).alias("asin_fba_fee"),
F.lit(None).alias("asin_freight_air_fee"), F.lit(None).alias("asin_freight_ocean_fee"),
F.lit(None).alias("asin_operate_fee"), F.lit(None).alias("asin_air_freight_gross_margin"),
F.lit(None).alias("asin_ocean_freight_gross_margin"), "asin_crawl_date",
F.lit(None).alias("asin_package_quantity"), F.lit(None).alias("asin_pattern_name"),
"category_first_id", "category_id", "first_category_rank", "current_category_rank",
"asin_weight_ratio", "asin_bought_month", F.lit(None).alias("buy_data_bought_week"),
F.lit(None).alias("buy_data_viewed_month"), F.lit(None).alias("buy_data_viewed_week"),
F.lit(None).alias("theme_en"), F.lit(None).alias("theme_label_en"), "asin_lqs_rating",
"asin_lqs_rating_detail", "title_matching_degree", "zr_flow_proportion", "matrix_flow_proportion",
"matrix_ao_val", "follow_sellers_count", "seller_json",
F.lit(self.site_name).alias("site_name"), F.lit(self.date_type).alias("date_type"),
F.lit(self.date_info).alias("date_info"))
self.df_save = self.df_save.na.fill(
{"asin_zr_counts": 0, "asin_sp_counts": 0, "asin_sb_counts": 0, "asin_vi_counts": 0, "asin_bs_counts": 0,
"asin_ac_counts": 0, "asin_tr_counts": 0, "asin_er_counts": 0, "asin_title_len": 0,
"asin_total_comments": 0, "variation_num": 0, "asin_img_num": 0, "act_one_two_val": 0.0,
"act_three_four_val": 0.0, "act_five_six_val": 0.0, "act_eight_val": 0.0, "one_star": 0,
"two_star": 0, "three_star": 0, "four_star": 0, "five_star": 0, "low_star": 0, "asin_size_type": 0,
"asin_rating_type": 0, "asin_site_name_type": 0, "asin_weight_type": 0, "asin_launch_time_type": 0,
"asin_ao_val_type": 0, "asin_rank_type": 0, "asin_price_type": 0, "asin_quantity_variation_type": 0,
"package_quantity": 1, "is_movie_label": 0, "is_brand_label": 0, "is_alarm_brand": 0,
"title_matching_degree": 0.0, "asin_lqs_rating": 0.0, "follow_sellers_count": -1})
self.df_save = self.df_save.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_save = self.df_save.drop_duplicates(['asin']).filter(F.length(F.col("asin"))<=10)
print("数据量为：", self.df_save.count())
self.df_save.show(10, truncate=False)
# 保存数据
def save_data(self):
print(f"清除hdfs目录中:{self.hdfs_path}")
HdfsUtils.delete_file_in_folder(self.hdfs_path)
partition_by = ["site_name", "date_type", "date_info"]
print(f"当前存储的表名为：{self.hive_tb},分区为{partition_by}", )
df_save = self.df_save
df_save = df_save.drop("seller_json")
df_save.write.saveAsTable(name=self.hive_tb, format='hive', mode='append', partitionBy=self.partitions_by)
print("save asin_detail success")
if self.date_type in ['month', 'month_week'] and self.date_info >= '2024-06':
max_report_sql = f"""
select max(report_date) as completed_date_info from workflow_everyday where site_name='us' and date_type='month' and page='流量选品' and status_val=14
"""
mysql_con = DBUtil.get_connection_info("mysql", "us")
df_date_info = SparkUtil.read_jdbc_query(session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
username=mysql_con['username'], query=max_report_sql)
completed_date_info = df_date_info.take(1)[0]['completed_date_info']
if self.date_info > completed_date_info:
print("往doris存储最新asin详情信息：")
df_doris = self.df_save.\
select("asin", "asin_ao_val", "asin_title", "asin_title_len", "asin_category_desc", "asin_volume",
"asin_weight", "asin_launch_time", "asin_brand_name", "one_star", "two_star", "three_star",
"four_star", "five_star", "low_star", "account_name", "account_id", "seller_country_name",
"category_first_id", "parent_asin", "variation_num", "img_info", "asin_crawl_date", "asin_price",
"asin_rating", "asin_total_comments", "matrix_ao_val", "zr_flow_proportion",
"matrix_flow_proportion", "date_info", F.col("asin_img_url").alias("img_url"),
F.col("category_id").alias("category_current_id"),
F.col("first_category_rank").alias("category_first_rank"),
F.col("current_category_rank").alias("category_current_rank"), "asin_type",
"bsr_orders", F.col("sales").alias("bsr_orders_sale"),
F.col("asin_page_inventory").alias("page_inventory"), "asin_bought_month", "seller_json",
F.col("asin_buy_box_seller_type").alias("buy_box_seller_type")
)
table_columns = """asin, asin_ao_val, asin_title, asin_title_len, asin_category_desc, asin_volume,
asin_weight, asin_launch_time, asin_brand_name, one_star, two_star, three_star, four_star, five_star, low_star,
account_name, account_id, seller_country_name, category_first_id, parent_asin, variation_num, img_info,
asin_crawl_date, asin_price, asin_rating, asin_total_comments, matrix_ao_val, zr_flow_proportion, matrix_flow_proportion,
date_info, img_url, category_current_id, category_first_rank, category_current_rank, asin_type, bsr_orders, bsr_orders_sale,
page_inventory, asin_bought_month, seller_json, buy_box_seller_type"""
DorisHelper.spark_export_with_columns(df_save=df_doris, db_name=self.doris_db, table_name=self.asin_latest_detail_table, table_columns=table_columns)
print("save asin_latest_detail success")
else:
print("不用导出旧数据到doris中")
pass
def handle_data(self):
self.handle_asin_basic_attribute()
self.handle_asin_detail_all_type()
self.handle_asin_category_info()
self.handle_asin_measure()
self.handle_seller_country()
self.handle_asin_lqs_rating()
self.handle_asin_is_hide()
self.handle_title_matching_degree()
self.handle_asin_attribute_change()
self.handle_column()
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 = DwtFlowAsin(site_name=site_name, date_type=date_type, date_info=date_info)
handle_obj.run()