import os
import sys
import time
import traceback
sys.path.append("/opt/module/spark-3.2.0-bin-hadoop3.2/demo/py_demo/")
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from utils.templates import Templates
from pyspark.sql import functions as F
from pyspark.sql.types import *
from utils.db_util import DBUtil
from utils.common_util import CommonUtil
from utils.spark_util import SparkUtil
from datetime import datetime, timedelta
from functools import reduce
from utils.es_util import EsUtils
from pyspark.sql import Window
from pyspark.storagelevel import StorageLevel
from utils.DorisHelper import DorisHelper
from yswg_utils.common_df import get_node_first_id_df, get_first_id_from_category_desc_df
from yswg_utils.common_udf import udf_parse_bs_category, parse_weight_str, udf_extract_volume_dimensions, udf_get_package_quantity_with_flag as udf_get_package_quantity, udf_parse_seller_json
class KafkaFlowAsinDetail(Templates):
def __init__(self, site_name='us', date_type="day", date_info='2022-10-01', consumer_type='latest', test_flag='normal', batch_size=100000):
super().__init__()
self.site_name = site_name
self.date_type = date_type
self.date_info = date_info
self.consumer_type = consumer_type # 消费实时还是消费历史
self.test_flag = test_flag # 正式环境跟测试环境
self.year = str(self.date_info).split('-')[0]
self.year_month = str(self.date_info).replace("-", "_")
self.repartition_num = 80
# kafka相关参数
self.topic_name = f"{self.site_name}_asin_detail_month_{self.year_month}"
self.batch_size = batch_size
self.schema = self.init_schema()
self.batch_size_history = 20000
self.processing_time = 900 if self.site_name == 'us' else 600
self.history_batch_id = 0
# doris相关参数
self.doris_db = "test" if self.test_flag == "test" else "selection"
self.max_bought_month_table = f"{self.site_name}_asin_max_bought_month_info"
self.parent_asin_latest_detail_table = f"{self.site_name}_parent_asin_latest_detail"
self.asin_latest_detail_table = f"{self.site_name}_asin_latest_detail"
# elasticsearch相关参数
self.client = EsUtils.get_es_client()
self.es_index_name = f"{self.topic_name}_test" if self.test_flag == 'test' else f"{self.topic_name}"
self.es_index_alias_name = f"{self.site_name}_st_detail_last_4_week_test" if self.test_flag == 'test' else f"{self.site_name}_st_detail_last_4_week"
self.es_index_body = EsUtils.get_es_body()
self.es_options = EsUtils.get_es_options(self.es_index_name)
self.db_save = 'kafka_flow_asin_detail'
self.app_name = self.get_app_name()
print(f"任务名称：{self.app_name}")
# Spark实时消费相关参数
self.spark = SparkUtil.get_stream_spark(app_name=self.app_name)
self.check_path = f"/tmp/wangrui/{self.topic_name}_{self.consumer_type}_test" if self.test_flag == 'test' else f"/tmp/wangrui/{self.topic_name}_{self.consumer_type}"
self.previous_date = self.get_previous_date(self)
self.previous_two_date = self.get_previous_two_date(self)
self.launch_time_interval_dict = self.get_launch_time_interval_dict()
print("日期字典：", self.launch_time_interval_dict)
self.initial_batch_id = self.get_initial_batch_id(self)
print("当前消费的起始批次为： ", self.initial_batch_id)
self.history_batch_id = self.initial_batch_id + 1
# BSR分类解析模板
self.pattern1_dict = {
"us": "See Top 100 in ".lower(),
"uk": "See Top 100 in ".lower(),
"de": "Siehe Top 100 in ".lower(),
"es": "Ver el Top 100 en ".lower(),
"fr": "Voir les 100 premiers en ".lower(),
"it": "Visualizza i Top 100 nella categoria ".lower(),
}
self.pattern_current_dict = {
"us": "#(\d+) ",
"uk": "(\d+) in ",
"de": "(\d+) in ",
"es": "(\d+) en ",
"fr": "(\d+) en ",
"it": "(\d+) in ",
}
# DataFrame初始化
self.df_previous_flow_asin = self.spark.sql("select 1+1;")
self.df_seller_info = self.spark.sql("select 1+1;")
self.df_self_asin_info = self.spark.sql("select 1+1;")
self.df_alarm_brand_info = self.spark.sql("select 1+1;")
self.df_asin_label_info = self.spark.sql("select 1+1;")
self.df_asin_measure = self.spark.sql("select 1+1;")
self.df_bs_report = self.spark.sql("select 1+1;")
self.df_asin_keep_date = self.spark.sql("select 1+1;")
self.df_asin_bsr_end = self.spark.sql("select 1+1;")
self.df_hide_category = self.spark.sql("select 1+1;")
self.df_asin_new_cate = self.spark.sql("select 1+ 1;")
self.df_user_package_num = self.spark.sql("select 1+1;")
self.df_asin_category = self.spark.sql("select 1+1;")
self.df_max_bought_month_info_update = self.spark.sql("select 1+1;")
# udf函数注册
package_schema = StructType([
StructField("parse_package_quantity", IntegerType(), True),
StructField("is_package_quantity_abnormal", IntegerType(), True),
])
self.u_parse_package_quantity = self.spark.udf.register('u_parse_package_quantity', udf_get_package_quantity, package_schema)
bs_category_schema = StructType([
StructField('asin_bs_cate_1_id', StringType(), True),
StructField('asin_bs_cate_current_id', StringType(), True),
StructField('asin_bs_cate_1_rank', IntegerType(), True),
StructField('asin_bs_cate_current_rank', IntegerType(), True),
])
self.u_parse_bs_category = self.spark.udf.register('u_parse_bs_category', udf_parse_bs_category, bs_category_schema)
weight_schema = StructType([
StructField('weight', FloatType(), True),
StructField('weight_type', StringType(), True)
])
self.u_parse_weight = self.spark.udf.register('u_parse_weight', parse_weight_str, weight_schema)
volume_schema = StructType([
StructField("length", FloatType(), True),
StructField("width", FloatType(), True),
StructField("height", FloatType(), True),
StructField("asin_volume_type", StringType(), True)
])
self.u_parse_volume = self.spark.udf.register('u_parse_volume', udf_extract_volume_dimensions, volume_schema)
seller_schema = StructType([
StructField("buy_box_seller_type", IntegerType(), True),
StructField("account_name", StringType(), True),
StructField("account_id", StringType(), True)
])
self.u_parse_seller_info = self.spark.udf.register('u_parse_seller_info', udf_parse_seller_json, seller_schema)
@staticmethod
def init_schema():
schema = StructType([
StructField("asin", StringType(), True),
StructField("title", StringType(), True),
StructField("img_url", StringType(), True),
StructField("rating", DoubleType(), True),
StructField("total_comments", IntegerType(), True),
StructField("price", FloatType(), True),
StructField("category", StringType(), True),
StructField("launch_time", StringType(), True),
StructField("volume", StringType(), True),
StructField("page_inventory", IntegerType(), True),
StructField("asin_vartion_list", ArrayType(ArrayType(StringType()), True), True),
StructField("title_len", IntegerType(), True),
StructField("img_num", IntegerType(), True),
StructField("img_type", StringType(), True),
StructField("activity_type", StringType(), True),
StructField("one_two_val", StringType(), True),
StructField("three_four_val", StringType(), True),
StructField("five_six_val", StringType(), True),
StructField("eight_val", StringType(), True),
StructField("node_id", StringType(), True),
StructField("five_star", IntegerType(), True),
StructField("four_star", IntegerType(), True),
StructField("three_star", IntegerType(), True),
StructField("two_star", IntegerType(), True),
StructField("one_star", IntegerType(), True),
StructField("low_star", IntegerType(), True),
StructField("together_asin", StringType(), True),
StructField("brand", StringType(), True),
StructField("ac_name", StringType(), True),
StructField("material", StringType(), True),
StructField("data_type", IntegerType(), True),
StructField("weight_str", StringType(), True),
StructField("seller_id", StringType(), True),
StructField("variat_num", IntegerType(), True),
StructField("best_sellers_rank", StringType(), True),
StructField("best_sellers_herf", StringType(), True),
StructField("account_name", StringType(), True),
StructField("parentAsin", StringType(), True),
StructField("asinUpdateTime", StringType(), True),
StructField("all_best_sellers_herf", StringType(), True),
StructField("image_view", IntegerType(), True),
StructField("product_description", StringType(), True),
StructField("describe", StringType(), True),
StructField("buy_sales", StringType(), True),
StructField("lob_asin_json", StringType(), True),
StructField("seller_json", StringType(), True),
StructField("customer_reviews_json", StringType(), True),
StructField("img_list", StringType(), True),
StructField("follow_sellers", IntegerType(), True)
])
return schema
@staticmethod
def get_previous_date(self):
self.df_date = self.spark.sql(f"select * from dim_date_20_to_30")
df = self.df_date.toPandas()
df_loc = df.loc[(df.year_month == f'{self.date_info}') & (df.day == 1)]
cur_month_id = int(list(df_loc.id)[0])
previous_date_id = cur_month_id - 1
df_loc = df.loc[df.id == previous_date_id]
previous_date = str(list(df_loc.year_month)[0])
return previous_date
@staticmethod
def get_previous_two_date(self):
self.df_date = self.spark.sql(f"select * from dim_date_20_to_30")
df = self.df_date.toPandas()
df_loc = df.loc[(df.year_month == f'{self.date_info}') & (df.day == 1)]
cur_month_id = int(list(df_loc.id)[0])
previous_two_date_id = cur_month_id - 40
df_loc = df.loc[df.id == previous_two_date_id]
prvious_two_date = str(list(df_loc.year_month)[0])
return prvious_two_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 get_initial_batch_id(self):
max_bought_month_batch_id_sql = f"""
SELECT MAX(batch_id) as initial_batch_id from {self.doris_db}.{self.max_bought_month_table} WHERE date_info='{self.date_info}' AND consumer_type='{self.consumer_type}'
"""
df_max_bought_month_batch_id = DorisHelper.spark_import_with_sql(
self.spark, query=max_bought_month_batch_id_sql)
max_bought_month_batch_id = 0 if df_max_bought_month_batch_id.take(1)[0]['initial_batch_id'] is None else \
df_max_bought_month_batch_id.take(1)[0]['initial_batch_id']
return max_bought_month_batch_id
# 1. 处理asin分类及排名以及排名类型字段
def handle_asin_bs_category_info(self, df):
df = df.withColumnRenamed("parentAsin", "parent_asin")
cate_current_pattern = self.pattern_current_dict[self.site_name]
cate_1_pattern = self.pattern1_dict[self.site_name]
df = df.withColumn("asin_bs_sellers_rank_lower", F.lower("best_sellers_rank"))
df = df.withColumn("asin_bs", self.u_parse_bs_category(
"asin_bs_sellers_rank_lower", "best_sellers_herf", "all_best_sellers_herf", F.lit(cate_current_pattern), F.lit(cate_1_pattern)))
df = df.withColumn("asin_bs_cate_1_id", df.asin_bs.getField("asin_bs_cate_1_id")) \
.withColumn("asin_bs_cate_current_id", df.asin_bs.getField("asin_bs_cate_current_id")) \
.withColumn("asin_bs_cate_1_rank", df.asin_bs.getField("asin_bs_cate_1_rank")) \
.withColumn("asin_bs_cate_current_rank", df.asin_bs.getField("asin_bs_cate_current_rank")) \
.drop("asin_bs", "asin_bs_sellers_rank_lower", "best_sellers_herf", "all_best_sellers_herf",
"best_sellers_rank")
df = df.withColumn("rank_type", F.expr("""
CASE WHEN asin_bs_cate_1_rank IS NOT NULL AND asin_bs_cate_1_rank BETWEEN 0 AND 1000 THEN 1
WHEN asin_bs_cate_1_rank BETWEEN 1000 AND 5000 THEN 2 WHEN asin_bs_cate_1_rank BETWEEN 5000 AND 10000 THEN 3
WHEN asin_bs_cate_1_rank BETWEEN 10000 AND 20000 THEN 4 WHEN asin_bs_cate_1_rank BETWEEN 20000 AND 30000 THEN 5
WHEN asin_bs_cate_1_rank BETWEEN 30000 AND 50000 THEN 6 WHEN asin_bs_cate_1_rank BETWEEN 50000 AND 70000 THEN 7
WHEN asin_bs_cate_1_rank >= 70000 THEN 8 ELSE 0 END"""))
return df
# 2. 利用node_id以及分类描述进行分类补充（此时无排名信息）
def handle_asin_category_supplement(self, df):
df = df.join(self.df_asin_new_cate, on=['node_id'], how='left')
df = df.withColumn("asin_bs_cate_current_id", F.coalesce(F.col("asin_bs_cate_current_id"), F.col("node_id"))). \
withColumn("asin_bs_cate_1_id", F.coalesce(F.col("asin_bs_cate_1_id"), F.col("category_first_id"))). \
drop("category_first_id", "node_id")
df_with_category = df.filter("asin_bs_cate_1_id is null and category is not null").select("asin", "category")
df_with_category = df_with_category.withColumn(
"category_split", F.split(F.col("category"), "›")
).withColumn(
"category_first_name", F.lower(F.col("category_split").getItem(0))
).drop("category_split", "category")
df_with_category = df_with_category.join(self.df_asin_category, on=['category_first_name'], how='inner')
df_with_category = df_with_category.withColumnRenamed("category_first_id", "category_first_id_with_name").drop("category_first_name")
df = df.join(df_with_category, on=['asin'], how='left')
df = df.withColumn("asin_bs_cate_1_id", F.coalesce(F.col("asin_bs_cate_1_id"), F.col("category_first_id_with_name"))).drop("category_first_id_with_name")
return df
# 3. 处理bsr销量、价格类型字段以及BSR销售额信息
def handle_asin_bsr_orders(self, df):
df = df.join(self.df_bs_report, on=['asin_bs_cate_1_id', 'asin_bs_cate_1_rank'], how='left')
df = df.withColumn("price_type", F.expr("""
CASE WHEN price IS NOT NULL AND price > 0 AND price < 10 THEN 1 WHEN price >= 10 AND price < 15 THEN 2
WHEN price >= 15 AND price < 20 THEN 3 WHEN price >= 20 AND price < 30 THEN 4
WHEN price >= 30 AND price < 50 THEN 5 WHEN price >= 50 THEN 6 ELSE 0 END""")).\
withColumn("bsr_orders_sale", F.round(F.col("bsr_orders") * F.col("price"), 2))
return df
# 4.解析Make-It-A-Bundle信息
def handle_asin_lob_info(self, df):
df = df.withColumn("is_contains_lob_info",
F.when(F.col("lob_asin_json").isNotNull(), F.lit(1)).otherwise(F.lit(0)))
df_parsed = df.withColumn("parse_asin_lob",
F.when(F.col("is_contains_lob_info") == 1, F.from_json("lob_asin_json", "array<struct<lob_asin:string>>")))
df_result = df_parsed.withColumn("asin_lob_info", F.expr("transform(parse_asin_lob, x -> x.lob_asin)"))
df = df_result.withColumn(
"asin_lob_info", F.regexp_replace(F.concat_ws(",", "asin_lob_info"), "[{}]", "")).drop(
"parse_asin_lob", "lob_asin_json")
return df
# 5. 处理配送方式、卖家所在地以及卖家所在地类型
def handle_asin_buy_box_seller_type(self, df):
df = df.withColumn("seller_json_parsed", self.u_parse_seller_info(df.seller_json))
df = df.withColumn("buy_box_seller_type", df.seller_json_parsed.buy_box_seller_type).withColumn(
"account_name", df.seller_json_parsed.account_name).drop("seller_json_parsed")
df = df.join(self.df_seller_info, on=['seller_id'], how='left')
df = df.withColumn("site_name_type", F.expr("""
CASE WHEN buy_box_seller_type = 1 THEN 4
WHEN buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%US%' THEN 1
WHEN buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%CN%' THEN 2
ELSE 3 END"""))
return df
# 6. 处理asin基础属性信息(长宽高重量等)
def handle_asin_basic_attribute_info(self, df):
# 1.解析ASIN重量相关信息
df = df.withColumn("weight_str", F.lower(F.col("weight_str"))).withColumn("asin_weight", self.u_parse_weight("weight_str", F.lit(self.site_name))).drop("weight_str")
df = df.withColumn(
"weight", F.when(df.asin_weight.getField("weight_type") == 'pounds', df.asin_weight.getField("weight")).otherwise(F.lit(0))).drop("asin_weight")
# 2.处理重量类型
df = df.withColumn("weight_type", F.expr("""
CASE WHEN weight BETWEEN 0 AND 0.2 THEN 1 WHEN weight BETWEEN 0.2 AND 0.4 THEN 2
WHEN weight BETWEEN 0.4 AND 0.6 THEN 3 WHEN weight BETWEEN 0.6 AND 1 THEN 4
WHEN weight BETWEEN 1 AND 2 THEN 5 WHEN weight >= 2 THEN 6 ELSE 0 END"""))
# 3.解析ASIN体积相关信息
df = df.withColumn("asin_volume", self.u_parse_volume("volume"))
df = df.withColumn("asin_volume_type", df.asin_volume.getField("asin_volume_type")) \
.withColumn("asin_length", F.when(F.col("asin_volume_type") == 'inches', df.asin_volume.getField("length"))) \
.withColumn("asin_width", F.when(F.col("asin_volume_type") == 'inches', df.asin_volume.getField("width"))) \
.withColumn("asin_height", F.when(F.col("asin_volume_type") == 'inches', df.asin_volume.getField("height"))) \
.drop("asin_volume", "asin_volume_type")
# 4.获取体积重/毛重相关信息
df = df.withColumn(
"asin_weight_ratio", F.when(
F.col("asin_length").isNotNull() & (F.col("asin_width").isNotNull()) &
(F.col("asin_height").isNotNull()) &
(F.col("weight") > 0), F.round(F.col("asin_length") * F.col("asin_width") * F.col("asin_height") * 3.2774128 / (F.col("weight") * 453.59), 3))
.otherwise(F.lit(-1)))
# 5.处理尺寸类型
if self.site_name == 'us':
expr_str = f"""
CASE WHEN weight > 0 AND 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 weight > 0 AND 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 weight > 0 AND 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 weight > 0 AND weight <= 150 AND asin_length > 0 AND asin_length <= 108 AND asin_length + asin_length + (asin_width + asin_height) * 2 <= 130 THEN 4
WHEN weight > 0 AND weight <= 150 AND asin_length > 0 AND asin_length <= 108 AND asin_length + asin_length + (asin_width + asin_height) * 2 <= 165 THEN 5
WHEN 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 weight > 0 AND 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 weight > 0 AND 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 weight > 0 AND 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 weight > 0 AND 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 weight > 0 AND 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"""
df = df.withColumn("size_type", F.expr(expr_str)).drop("asin_length", "asin_width", "asin_height")
return df
# 7. 处理asin图片信息
def handle_asin_img_info(self, df):
img_schema = ArrayType(ArrayType(StringType()))
df = df.withColumn("img_list", F.from_json(F.col("img_list"), img_schema))
df_with_img = df.filter(F.size("img_list") > 0).select("asin", "img_list")
df_with_img_attribute = df_with_img.select(
"asin", F.explode("img_list").alias("img_attributes")
).select(
"asin", F.col("img_attributes")[1].alias("img_url"), F.col("img_attributes")[2].alias("img_order_by"),
F.col("img_attributes")[3].alias("data_type")
)
df_with_img_attribute_agg = df_with_img_attribute.groupby("asin").agg(
F.to_json(F.collect_list(F.struct(F.col("img_url"), F.col("img_order_by"), F.col("data_type")))).alias(
"img_info")
)
df = df.drop("img_list")
df = df.join(df_with_img_attribute_agg, on=['asin'], how='left')
return df
# 8. 处理变体相关（ao及母体相关，自然占比及母体自然占比，各类型数量，月销信息等）
def handle_asin_measure(self, df):
df = CommonUtil.get_asin_variant_attribute(df_asin_detail=df, df_asin_measure=self.df_asin_measure,
partition_num=self.repartition_num, use_type=1)
# 是否数量变体类型和ao的类型
df = df.withColumn("quantity_variation_type", F.expr("""
CASE WHEN size is not null and size != '' and lower(size) like '%quantity%' THEN 1 ELSE 0 END""")).withColumn(
"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"""))
df = df.withColumnRenamed("asin_zr_counts", "zr_counts").withColumnRenamed("asin_ao_val", "ao_val") \
.withColumnRenamed("asin_zr_flow_proportion", "zr_flow_proportion") \
.withColumnRenamed("asin_amazon_orders", "asin_bought_month").drop("asin_st_counts", "asin_adv_counts")
# 获取parent_asin下最新ASIN信息
df_parent_asin_info = df.filter("parent_asin is not null").select("parent_asin", "asin_vartion_list", "asinUpdateTime")
parent_asin_window = Window.partitionBy(['parent_asin']).orderBy(
F.desc_nulls_last("asinUpdateTime")
)
df_parent_asin_info = df_parent_asin_info.withColumn("u_rank", F.row_number().over(window=parent_asin_window))
df_parent_asin_info = df_parent_asin_info.repartition(self.repartition_num)
df_parent_asin_info = df_parent_asin_info.filter("u_rank = 1").drop("u_rank")
df_asin_variat = df_parent_asin_info.filter(F.size("asin_vartion_list") > 0).\
select("parent_asin", "asinUpdateTime", F.explode("asin_vartion_list").alias("variant_attribute")).\
select("parent_asin", F.col("asinUpdateTime").alias("asin_crawl_date"),
F.col("variant_attribute")[0].alias("asin"), F.col("variant_attribute")[1].alias("color"),
F.col("variant_attribute")[3].alias("size"), F.col("variant_attribute")[5].alias("style"))
df_asin_variat_agg = df_asin_variat.groupby(['parent_asin']).agg(
F.first("asin_crawl_date").alias("asin_crawl_date"),
F.concat_ws(',', F.collect_list("asin")).alias("variation_info"),
F.to_json(F.collect_list(F.struct(F.col("color"), F.col("size"), F.col("style")))).alias("attr_info")
)
print("导出父ASIN最新变体信息到doris：")
df_doris = df_asin_variat_agg.select(
"parent_asin", F.lit(self.date_info).alias("date_info"), "asin_crawl_date", "variation_info", "attr_info")
table_columns = "parent_asin, date_info, asin_crawl_date, variation_info, attr_info"
DorisHelper.spark_export_with_columns(df_save=df_doris, db_name=self.doris_db, table_name=self.parent_asin_latest_detail_table, table_columns=table_columns)
df_doris.unpersist()
return df
# 9. 提取打包数量字段
def handle_asin_package_quantity(self, df):
df = df.withColumn(
"variat_attribute", F.concat_ws("&&&%", F.col("color"), F.col("style"), F.col("size"), F.col("material")))
df = df.withColumn("title_parse", self.u_parse_package_quantity(df.title)).withColumn(
"variat_parse", self.u_parse_package_quantity(df.variat_attribute))
df = df.withColumn("title_package_quantity", df.title_parse.getField("parse_package_quantity")). \
withColumn("variat_package_quantity", df.variat_parse.getField("parse_package_quantity")). \
withColumn("title_package_quantity_is_abnormal", df.title_parse.getField("is_package_quantity_abnormal")). \
withColumn("variat_package_quantity_is_abnormal", df.variat_parse.getField("is_package_quantity_abnormal")). \
drop("title_parse", "variat_parse", "variat_attribute")
df = df.withColumn(
"package_quantity", F.expr("""
CASE WHEN title_package_quantity is null and variat_package_quantity is not null THEN variat_package_quantity
WHEN title_package_quantity is not null THEN title_package_quantity ELSE 1 END""")
).withColumn(
"is_package_quantity_abnormal", F.expr("""
CASE WHEN title_package_quantity is null and variat_package_quantity is not null THEN variat_package_quantity_is_abnormal
WHEN title_package_quantity is not null THEN title_package_quantity_is_abnormal ELSE 2 END""")
).drop("title_package_quantity", "variat_package_quantity", "title_package_quantity_is_abnormal", "variat_package_quantity_is_abnormal")
df = df.withColumn("title", F.lower(F.col("title")))
df = df.join(self.df_user_package_num, on=['asin', 'title'], how='left')
df = df.withColumn("package_quantity", F.coalesce(F.col("user_package_num"), F.col("package_quantity"))). \
withColumn(
"is_package_quantity_abnormal", F.coalesce(F.col("user_is_package_quantity_abnormal"), F.col("is_package_quantity_abnormal"))
).drop("user_package_num", "user_is_package_quantity_abnormal")
return df
# 10. 处理品牌标签、是否告警品牌、处理asin_lqs_rating信息
def handle_asin_lqs_and_brand(self, df):
# 1.品牌标签以及是否告警品牌
df = df.withColumn("is_brand_label", F.expr("""CASE WHEN brand is not null THEN 1 ELSE 0 END"""))
df = df.withColumn("brand", F.lower("brand"))
df = df.join(self.df_alarm_brand_info, on=['brand'], how='left')
df = df.withColumn("is_alarm_brand",
F.when(F.col("is_alarm_brand").isNotNull(), F.col("is_alarm_brand")).otherwise(F.lit(0)))
# 2. lqs评分
df = df.withColumn("category_node_rating",
F.expr(f"""CASE WHEN asin_bs_cate_current_id is not null THEN 1 ELSE 0 END""")) \
.withColumn("zr_rating", F.expr(f"""CASE WHEN zr_counts > 0 THEN 0.5 ELSE 0 END""")) \
.withColumn("sp_rating", F.expr(f"""CASE WHEN sp_counts > 0 THEN 1 ELSE 0 END""")) \
.withColumn("a_add_rating", F.expr(f"""CASE WHEN img_type like '%3%' THEN 1 ELSE 0 END""")) \
.withColumn("video_rating", F.expr(f"""CASE WHEN img_type like '%2%' THEN 0.5 ELSE 0 END""")) \
.withColumn("brand_rating", F.expr(f"""CASE WHEN is_brand_label = 1 THEN 0.2 ELSE 0 END""")) \
.withColumn("product_describe_rating",
F.expr(f"""CASE WHEN product_description is not null THEN 0.2 ELSE 0 END""")) \
.withColumn("highlight_rating", F.expr(f"""
CASE WHEN describe is not null AND size(split(describe, '\\|-\\|')) <= 4 THEN size(split(describe, '\\|-\\|')) * 0.4
WHEN describe is not null AND size(split(describe, '\\|-\\|')) > 4 THEN 1.6 ELSE 0 END""")) \
.withColumn("title_len_rating", F.expr(f"""CASE WHEN title_len >= 50 AND title_len <=200 THEN 0.5 ELSE 0 END""")) \
.withColumn("title_brand_rating", F.expr(f"""
CASE WHEN brand is not null AND lower(regexp_replace(title, '[^a-zA-Z0-9\\s]', '')) LIKE lower(regexp_replace(brand, '[^a-zA-Z0-9\\s]', '')) || '%' THEN 0.5
ELSE 0 END""")) \
.withColumn("img_num_rating", F.expr(f"""
CASE WHEN img_num <= 4 THEN img_num * 0.5 WHEN img_num >4 THEN 2 ELSE 0 END""")) \
.withColumn("img_enlarge_rating", F.expr(f"""CASE WHEN image_view = 1 THEN 0.5 ELSE 0 END"""))
df = df.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")).cast("double")).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")))
)
df = df.drop("product_description", "describe", "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")
return df
# 11. 通过ASIN页面信息处理(评分类型、上架时间类型、电影标签、是否内部asin、是否隐藏分类、有效类型、必需ASIN、asin_type)
def handle_asin_detail_all_type(self, df):
# 1. 评分类型
df = df.withColumn("rating_type", F.expr("""
CASE WHEN rating >= 4.5 THEN 1 WHEN rating >= 4 AND rating < 4.5 THEN 2 WHEN rating >= 3.5 AND rating < 4 THEN 3
WHEN rating >= 3 AND rating < 3.5 THEN 4 WHEN rating < 3 AND rating >= 0 THEN 5 ELSE 0 END"""))
# 2. 上架时间类型
df = df.join(self.df_asin_keep_date, on=['asin'], how='left')
df = df.withColumn("launch_time", F.when(F.col("launch_time").isNull(), F.col("new_launch_time")).otherwise(
F.col("launch_time")))
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 launch_time >= '{one_month}' THEN 1
WHEN launch_time >= '{three_month}' AND launch_time < '{one_month}' THEN 2
WHEN launch_time >= '{six_month}' AND launch_time < '{three_month}' THEN 3
WHEN launch_time >= '{twelve_month}' AND launch_time < '{six_month}' THEN 4
WHEN launch_time >= '{twenty_four_month}' AND launch_time < '{twelve_month}' THEN 5
WHEN launch_time >= '{thirty_six_month}' AND launch_time < '{twenty_four_month}' THEN 6
WHEN launch_time < '{thirty_six_month}' THEN 7 ELSE 0 END"""
df = df.withColumn("launch_time_type", F.expr(expr_str))
# 3. 电影标签
movie_label_list = ['prime video', 'dvd', 'blu-ray', 'kindle', 'app', 'paperback', 'audible audiobook',
'kindle edition', 'kindle & comixology', 'hardcover', 'comic', 'multi-format', '4k',
'library binding', 'vinyl', 'audio cd', 'mp3 music', 'single issue magazine',
'print magazine', 'unknown binding']
df = df.join(self.df_asin_label_info, on=['asin'], how='left')
condition = reduce(
lambda acc, keyword: acc | F.expr(f"exists(asin_label_list, x -> x like '%{keyword}%')"),
movie_label_list,
F.lit(False)
)
df = df.withColumn("is_movie_label", condition.cast("int")).drop("asin_label_list")
# 4. 是否内部asin、是否隐藏分类
df = df.join(self.df_self_asin_info, on=['asin'], how='left')
df = df.withColumn(
"is_self_asin", F.when(F.col("is_self_asin").isNotNull(), F.col("is_self_asin")).otherwise(F.lit(0)))
df = df.join(self.df_hide_category, on=['asin_bs_cate_current_id'], how='left')
df = df.na.fill({"hide_flag": 0})
df = df.withColumn("is_hide_asin", F.expr("""
CASE WHEN hide_flag = 1 THEN 1 WHEN asin_bs_cate_1_id = 'grocery' and asin_bs_cate_current_id != '6492272011' THEN 1
WHEN asin_bs_cate_current_id in ('21393128011', '21377129011', '21377127011', '21377130011', '21388218011', '21377132011') THEN 1
ELSE 0 END""")).drop("hide_flag")
# 5. 有效类型
df = df.join(self.df_asin_bsr_end, on=['asin_bs_cate_1_id'], how='left')
df = df.withColumn("bsr_type", F.expr("""
CASE WHEN limit_rank is null and asin_bs_cate_1_rank <= 500000 THEN 1 WHEN limit_rank is not null and asin_bs_cate_1_rank <= limit_rank THEN 1 ELSE 0 END"""
)).drop("limit_rank")
# 5. 是否必需ASIN
df = df.withColumn("is_need_asin", F.expr("""
CASE WHEN asin_bs_cate_1_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"""))
# 6. asin_type
df = df.withColumn("asin_type", F.expr("""
CASE WHEN is_self_asin=1 THEN 1 WHEN is_need_asin=1 THEN 2 WHEN is_hide_asin=1 THEN 3 ELSE 0 END"""
)).drop("is_self_asin", "is_need_asin", "is_hide_asin")
return df
# 12. 处理变化率相关字段
def handle_asin_attribute_change(self, df):
# 处理ASIN维度的变化率信息
df = df.join(self.df_previous_flow_asin, on=['asin'], how='left')
columns_to_change = [
("ao_val", "previous_asin_ao_val", "ao"),
("price", "previous_asin_price", "price"),
("asin_bs_cate_1_rank", "previous_first_category_rank", "rank"),
("bsr_orders", "previous_asin_bsr_orders", "bsr_orders"),
("rating", "previous_asin_rating", "rating"),
("total_comments", "previous_asin_total_comments", "comments"),
("variat_num", "previous_asin_variation_num", "variation"),
("bsr_orders_sale", "previous_sales", "sales")
]
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
for current_col, previous_col, suffix in columns_to_change:
rise_col, change_col = calculate_change(current_col, previous_col)
if suffix == 'ao':
df = df.withColumn(f"{suffix}_rise", F.round(rise_col, 3))
elif suffix in ['price', 'sales']:
df = df.withColumn(f"{suffix}_rise", F.round(rise_col, 2))
elif suffix == 'rating':
df = df.withColumn(f"{suffix}_rise", F.round(rise_col, 1))
else:
df = df.withColumn(f"{suffix}_rise", rise_col.cast(IntegerType()))
df = df.withColumn(f"{suffix}_change", F.round(change_col, 4))
df = df.drop(previous_col)
return df
# 13. 字段标准化
def handle_column_name(self, df):
df = df.withColumnRenamed("asin_bs_cate_1_id", "category_first_id")\
.withColumnRenamed("asin_bs_cate_current_id", "category_id") \
.withColumnRenamed("asin_bs_cate_1_rank", "first_category_rank")\
.withColumnRenamed("asin_bs_cate_current_rank", "current_category_rank") \
.withColumnRenamed("variat_num", "variation_num")\
.withColumnRenamed("seller_id", "account_id").withColumnRenamed("seller_country_name", "site_name") \
.withColumnRenamed("asinUpdateTime", "asin_crawl_date")\
.withColumnRenamed("customer_reviews_json", "product_features")\
.withColumn("collapse_asin", F.coalesce(F.col("parent_asin"), F.col("asin")))\
.withColumn("bsr_best_orders_type", F.lit(-1))
df_save = df.select("asin", "ao_val", "zr_counts", "sp_counts", "sb_counts", "vi_counts", "bs_counts", "ac_counts",
"tr_counts", "er_counts", "bsr_orders", "bsr_orders_sale", "title", "title_len", "price",
"rating", "total_comments", "buy_box_seller_type", "page_inventory", "volume", "weight", "color",
"size", "style", "material", "launch_time", "img_num", "parent_asin", "img_type", "img_url",
"activity_type", "one_two_val", "three_four_val", "five_six_val", "eight_val", "brand",
"variation_num", "one_star", "two_star", "three_star", "four_star", "five_star", "low_star",
"together_asin", "account_name", "account_id", "rank_rise", "rank_change", "ao_rise",
"ao_change", "price_rise", "price_change", "rating_rise", "rating_change", "comments_rise",
"comments_change", "bsr_orders_rise", "bsr_orders_change", "sales_rise", "sales_change",
"variation_rise", "variation_change", "size_type", "rating_type", "site_name_type",
"weight_type", "launch_time_type", "ao_val_type", "rank_type", "price_type", "bsr_type",
"bsr_best_orders_type", "quantity_variation_type", "package_quantity", "is_movie_label",
"is_brand_label", "is_alarm_brand", "asin_type", "asin_crawl_date", "category_first_id",
"category_id", "first_category_rank", "current_category_rank", "asin_weight_ratio",
"site_name", "asin_bought_month", "asin_lqs_rating", "asin_lqs_rating_detail",
"asin_lob_info", "is_contains_lob_info", "is_package_quantity_abnormal", "category",
"zr_flow_proportion", "matrix_flow_proportion", "matrix_ao_val", "product_features", "img_info",
"collapse_asin", F.col("follow_sellers").alias("follow_sellers_count"), "seller_json")
df_save = df_save.na.fill(
{"zr_counts": 0, "sp_counts": 0, "sb_counts": 0, "vi_counts": 0, "bs_counts": 0, "ac_counts": 0,
"tr_counts": 0, "er_counts": 0, "title_len": 0, "total_comments": 0, "variation_num": 0, "img_num": 0,
"one_two_val": 0.0, "three_four_val": 0.0, "five_six_val": 0.0, "eight_val": 0.0,
"one_star": 0, "two_star": 0, "three_star": 0, "four_star": 0, "five_star": 0, "low_star": 0,
"size_type": 0, "rating_type": 0, "site_name_type": 0, "weight_type": 0, "launch_time_type": 0,
"ao_val_type": 0, "rank_type": 0, "price_type": 0, "quantity_variation_type": 0, "package_quantity": 1,
"is_movie_label": 0, "is_brand_label": 0, "is_alarm_brand": 0, "asin_lqs_rating": 0.0, "follow_sellers_count": -1}
)
print("asin的标准信息：")
df_save.show(10, truncate=False)
return df_save
def read_data(self):
print("1. 读取上个维度的flow_asin")
sql = f"""
select asin, asin_ao_val as previous_asin_ao_val, asin_price as previous_asin_price,
variation_num as previous_asin_variation_num, asin_rating as previous_asin_rating,
asin_total_comments as previous_asin_total_comments, first_category_rank as previous_first_category_rank,
bsr_orders as previous_asin_bsr_orders, sales as previous_sales
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_previous_flow_asin = self.spark.sql(sqlQuery=sql)
if self.df_previous_flow_asin.count() <= 1:
print("该历史节点数据不全，调整到上上个月")
sql = f"""
select asin, first_category_rank as previous_first_category_rank,
round(asin_ao_val, 3) as previous_asin_ao_val, asin_price as previous_asin_price,
bsr_orders as previous_bsr_orders, asin_rating as previous_asin_rating,
asin_total_comments as previous_asin_total_comments, sales as previous_sales,
variation_num as previous_variation_num
from dwt_flow_asin where site_name = '{self.site_name}' and date_type = '{self.date_type}'
and date_info = '{self.previous_two_date}'
"""
print("sql=", sql)
self.df_previous_flow_asin = self.spark.sql(sqlQuery=sql)
self.df_previous_flow_asin = self.df_previous_flow_asin.repartition(self.repartition_num).persist(StorageLevel.DISK_ONLY)
self.df_previous_flow_asin.show(10, truncate=False)
print("2. 获取卖家相关信息")
sql = f"""
select fd_unique as seller_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"""
print("sql=", sql)
self.df_seller_info = self.spark.sql(sqlQuery=sql)
self.df_seller_info = self.df_seller_info.repartition(self.repartition_num).persist(StorageLevel.DISK_ONLY)
self.df_seller_info.show(10, truncate=False)
print("3. 读取内部asin信息")
sql = f"""select asin, 1 as is_self_asin from {self.site_name}_self_asin group by asin"""
print("sql=", sql)
mysql_con_info = DBUtil.get_connection_info(db_type='mysql', site_name=self.site_name)
if mysql_con_info is not None:
df_self_asin_info = SparkUtil.read_jdbc_query(
session=self.spark, url=mysql_con_info['url'], pwd=mysql_con_info['pwd'],
username=mysql_con_info['username'], query=sql)
self.df_self_asin_info = F.broadcast(df_self_asin_info)
self.df_self_asin_info.show(10, truncate=False)
print("4. 读取告警品牌信息")
sql = f"""
select brand, 1 as is_alarm_brand
from (select lower(trim(brand_name)) as brand
from brand_alert_erp where brand_name is not null) t group by brand"""
print("sql=", sql)
if self.site_name == 'us':
pg_cluster_con_info = DBUtil.get_connection_info(db_type="postgresql_cluster", site_name=self.site_name)
if pg_cluster_con_info is not None:
df_alarm_brand_info = SparkUtil.read_jdbc_query(
session=self.spark, url=pg_cluster_con_info['url'], pwd=pg_cluster_con_info['pwd'],
username=pg_cluster_con_info['username'], query=sql)
self.df_alarm_brand_info = F.broadcast(df_alarm_brand_info)
self.df_alarm_brand_info.show(10, truncate=False)
else:
schema = StructType([
StructField("brand", StringType(), True),
StructField("is_alarm_brand", IntegerType(), True)
])
self.df_alarm_brand_info = self.spark.createDataFrame([], schema)
print("5. 读取隐藏分类信息")
sql = f"""
select category_id_base as asin_bs_cate_current_id, 1 as hide_flag from us_bs_category_hide group by category_id_base
"""
print("sql=", sql)
us_mysql_con_info = DBUtil.get_connection_info(db_type='mysql', site_name='us')
if us_mysql_con_info is not None:
df_hide_category = SparkUtil.read_jdbc_query(
session=self.spark, url=us_mysql_con_info['url'], pwd=us_mysql_con_info['pwd'],
username=us_mysql_con_info['username'], query=sql)
self.df_hide_category = F.broadcast(df_hide_category)
self.df_hide_category.show(10, truncate=False)
print("6. 读取asin_label信息")
sql = f"""
select asin, label from
(select asin, lower(label) as label, created_time,row_number() over(partition by asin,label order by updated_time desc) as crank
from ods_other_search_term_data where site_name='{self.site_name}' and date_type='{self.date_type}' and
date_info='{self.date_info}' and trim(label) not in ('null','') and label is not null) t where t.crank=1
"""
print("sql=", sql)
self.df_asin_label_info = self.spark.sql(sqlQuery=sql)
if self.df_asin_label_info.count() <= 1:
print("该历史节点数据不全，调整到上上个月")
sql = f"""
select asin, label from
(select asin, lower(label) as label, created_time,row_number()
over(partition by asin,label order by updated_time desc) as crank
from ods_other_search_term_data where site_name='{self.site_name}' and date_type='{self.date_type}' and
date_info='{self.previous_date}' and trim(label) not in ('null','') and label is not null) t where t.crank=1
"""
print("sql=", sql)
self.df_asin_label_info = self.spark.sql(sqlQuery=sql)
self.df_asin_label_info = self.df_asin_label_info.groupby(['asin']).agg(
F.collect_set("label").alias("asin_label_list"))
self.df_asin_label_info = self.df_asin_label_info.repartition(self.repartition_num).persist(StorageLevel.DISK_ONLY)
self.df_asin_label_info.show(10, truncate=False)
print("7. 读取dwd_asin_measure拿取ao及各类型数量")
sql = f"""
select asin, asin_sp_counts as sp_counts, (asin_sb1_counts + asin_sb2_counts) as sb_counts, asin_sb3_counts as vi_counts,
asin_bs_counts as bs_counts, asin_ac_counts as ac_counts, asin_tr_counts as tr_counts, asin_er_counts as er_counts,
asin_st_counts, asin_zr_counts, asin_adv_counts, round(asin_zr_flow_proportion, 3) as asin_zr_flow_proportion,
round(asin_ao_val, 3) as asin_ao_val, asin_amazon_orders
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(self.repartition_num).persist(StorageLevel.DISK_ONLY)
self.df_asin_measure.show(10, truncate=False)
print("8. 读取one_category_report表")
if int(self.year) == 2022 and int(self.month) < 3:
sql = f"select category_id as asin_bs_cate_1_id, rank as asin_bs_cate_1_rank, orders as bsr_orders from ods_one_category_report " \
f"where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='2022-12'"
else:
sql = f"select category_id as asin_bs_cate_1_id, rank as asin_bs_cate_1_rank, orders as bsr_orders from ods_one_category_report " \
f"where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"
print("sql=", sql)
self.df_bs_report = self.spark.sql(sqlQuery=sql)
self.df_bs_report = self.df_bs_report.repartition(self.repartition_num).persist(StorageLevel.DISK_ONLY)
self.df_bs_report.show(10, truncate=False)
print("9. 读取keep_date获取上架时间")
sql = f"""
select asin, new_launch_time from
(select asin, launch_time as new_launch_time,
row_number() over(partition by asin order by updated_at desc) as trank
from ods_asin_keep_date where site_name='{self.site_name}' and state=3) t where t.trank=1
"""
print("sql=", sql)
self.df_asin_keep_date = self.spark.sql(sqlQuery=sql)
self.df_asin_keep_date = self.df_asin_keep_date.repartition(self.repartition_num).persist(StorageLevel.DISK_ONLY)
self.df_asin_keep_date.show(10, truncate=False)
print("10. 读取bsr有效排名信息")
sql = f"""select rank as limit_rank, category_id as asin_bs_cate_1_id from {self.site_name}_bsr_end"""
print("sql=", sql)
if mysql_con_info is not None:
df_asin_bsr_end = SparkUtil.read_jdbc_query(
session=self.spark, url=mysql_con_info['url'], pwd=mysql_con_info['pwd'],
username=mysql_con_info['username'], query=sql)
self.df_asin_bsr_end = F.broadcast(df_asin_bsr_end)
self.df_asin_bsr_end.show(10, truncate=False)
print("11. 通过node_id获取一级分类进行分类补充")
df_asin_new_cate = get_node_first_id_df(self.site_name, self.spark)
self.df_asin_new_cate = F.broadcast(df_asin_new_cate)
self.df_asin_new_cate.show(10, truncate=False)
print("12. 获取用户修改打包数量信息")
pg_con_info = DBUtil.get_connection_info("postgresql", "us")
sql = f"""
WITH ranked_edit_logs AS (SELECT edit_key_id, lower(val_related_info) as val_related_info, val_after,
ROW_NUMBER() OVER (PARTITION BY edit_key_id ORDER BY create_time DESC) AS rn FROM sys_edit_log
WHERE module = '流量选品' AND filed = 'package_quantity' AND site_name='{self.site_name}')
SELECT edit_key_id as asin, val_related_info as title, cast(val_after as int) as user_package_num,
0 as user_is_package_quantity_abnormal FROM ranked_edit_logs WHERE rn = 1"""
if pg_con_info is not None:
df_user_package_num = SparkUtil.read_jdbc_query(
session=self.spark, url=pg_con_info['url'], pwd=pg_con_info['pwd'],
username=pg_con_info['username'], query=sql)
self.df_user_package_num = F.broadcast(df_user_package_num)
self.df_user_package_num.show(10, truncate=False)
print("14. 获取分类ID与分类名称的对应关系")
self.df_asin_category = get_first_id_from_category_desc_df(self.site_name, self.spark)
self.df_asin_category = self.df_asin_category.withColumn(
"category_first_name", F.lower("category_first_name")
).repartition(self.repartition_num).persist(StorageLevel.DISK_ONLY)
self.df_asin_category.show(10, truncate=False)
# 字段处理逻辑综合
def handle_all_field(self, df):
# 1. 处理asin分类及排名以及排名类型字段
df = self.handle_asin_bs_category_info(df)
# 2. 利用node_id进行分类补充
df = self.handle_asin_category_supplement(df)
# 3. 处理bsr销量及销售额信息以及价格类型字段
df = self.handle_asin_bsr_orders(df)
# 4. 解析Make-It-A-Bundle信息
df = self.handle_asin_lob_info(df)
# 5. 处理配送方式、卖家所在地以及卖家所在地类型
df = self.handle_asin_buy_box_seller_type(df)
# 6. 处理asin基础属性信息(长宽高重量等)
df = self.handle_asin_basic_attribute_info(df)
# 7. 处理asin图片信息
df = self.handle_asin_img_info(df)
# 8. 处理变体相关（ao及母体相关，自然占比及母体自然占比，各类型数量，月销信息等）
df = self.handle_asin_measure(df)
# 9. 提取打包数量字段
df = self.handle_asin_package_quantity(df)
# 10. 处理品牌标签、是否告警品牌、处理asin_lqs_rating信息
df = self.handle_asin_lqs_and_brand(df)
# 11.通过ASIN页面信息处理(评分类型、上架时间类型、电影标签、ASIN类型、有效类型)
df = self.handle_asin_detail_all_type(df)
# 12. 处理变化率相关字段
df = self.handle_asin_attribute_change(df)
# 13. 字段标准化
df_save = self.handle_column_name(df)
return df_save
# 写入es前的准备工作
def es_prepare(self):
# 创建对应es索引
EsUtils.create_index(self.es_index_name, self.client, self.es_index_body)
print("索引名称为：", self.es_index_name)
if not EsUtils.exist_index_alias(self.es_index_alias_name, self.client):
EsUtils.create_index_alias(self.es_index_name, self.es_index_alias_name, self.client)
else:
index_name_list = EsUtils.get_index_names_associated_alias(self.es_index_alias_name, self.client)
if self.es_index_name not in index_name_list:
EsUtils.delete_index_alias(self.es_index_alias_name, self.client)
EsUtils.create_index_alias(self.es_index_name, self.es_index_alias_name, self.client)
else:
pass
# 写入elasticsearch逻辑
def save_to_es(self, df, batch_num):
print("插入当前批次数据, 插入的数量量为： " + str(batch_num))
start_time = time.time()
df_asin_latest_detail = df. \
select("asin", F.col("ao_val").alias("asin_ao_val"), F.col("title").alias("asin_title"),
F.col("title_len").alias("asin_title_len"), F.col("category").alias("asin_category_desc"),
F.col("volume").alias("asin_volume"), F.col("weight").alias("asin_weight"),
F.col("launch_time").alias("asin_launch_time"), F.col("brand").alias("asin_brand_name"), "one_star",
"two_star", "three_star", "four_star", "five_star", "low_star", "account_name", "account_id",
F.col("site_name").alias("seller_country_name"), "category_first_id", "parent_asin", "variation_num",
"img_info", "asin_crawl_date", F.col("price").alias("asin_price"),
F.col("rating").alias("asin_rating"),
F.col("total_comments").alias("asin_total_comments"), "matrix_ao_val", "zr_flow_proportion",
"matrix_flow_proportion", F.lit(self.date_info).alias("date_info"), "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", "bsr_orders_sale", "page_inventory", "asin_bought_month", "seller_json", "buy_box_seller_type")
df = df.drop("category", "seller_json")
df.write.format("org.elasticsearch.spark.sql").options(**self.es_options).mode("append").save()
end_time = time.time()
elapsed_time = end_time - start_time
print("当前插入时长为：" + str(elapsed_time))
# ASIN最新详情的信息
if self.consumer_type == 'latest' and self.test_flag == 'normal':
print("导出ASIN最新详情信息到doris：")
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_asin_latest_detail, db_name=self.doris_db, table_name=self.asin_latest_detail_table, table_columns=table_columns)
df_asin_latest_detail.unpersist()
# 实时消费中批次数据的处理逻辑
def handle_kafka_stream(self, df, batch_id):
try:
batch_num = df.count()
if batch_num > 0:
start_time = time.time()
print("当前批次:" + str(batch_id) + "; 该批次数据量为：" + str(batch_num))
df = df.repartition(self.repartition_num)
batch_id = int(batch_id) + self.initial_batch_id
df_save = self.handle_all_field(df)
self.es_prepare()
self.save_to_es(df_save, batch_num)
df_save.unpersist()
end_time = time.time()
print("当前批次:" + str(batch_id) + "执行完毕, 执行时长为：" + str(end_time - start_time))
else:
print("当前批次没有数据")
except Exception as e:
print(e, traceback.format_exc())
# 消费主题下的所有历史数据
def handle_kafka_history(self, kafka_df):
print("处理kafka历史数据")
batch_num = kafka_df.count()
if batch_num > 0:
self.history_batch_id = self.history_batch_id + 1
start_time = time.time()
kafka_df = kafka_df.repartition(self.repartition_num)
kafka_df = self.handle_all_field(kafka_df)
self.es_prepare()
self.save_to_es(kafka_df, batch_num)
end_time = time.time()
print("该批次数据处理完毕, 执行时长为：" + str(end_time - start_time))
else:
raise ValueError("当前主题中没有数据,请注意检查！")
if __name__ == '__main__':
arguments = sys.argv[1:]
site_name = sys.argv[1] # 参数1：站点
date_type = sys.argv[2] # 参数2：类型：week/4_week/month/quarter/day
date_info = sys.argv[3] # 参数3：年-周/年-月/年-季/年-月-日, 比如: 2022-1
consumer_type = sys.argv[4] # 参数3：年-周/年-月/年-季/年-月-日, 比如: 2022-1
if len(arguments) == 5:
test_flag = sys.argv[5]
else:
test_flag = 'normal'
handle_obj = KafkaFlowAsinDetail(site_name=site_name, date_type=date_type, date_info=date_info, consumer_type=consumer_type, test_flag=test_flag, batch_size=200000)
handle_obj.run_kafka()