dwt_merchantwords_st_detail.py

import os
import sys

sys.path.append(os.path.dirname(sys.path[0]))
from utils.common_util import CommonUtil

from utils.spark_util import SparkUtil
from pyspark.sql import functions as F, Window, DataFrame
from pyspark.sql.types import IntegerType, StringType, MapType

"""
merchantwords 搜索词数据集成并去重输出
"""


class DwtMerchantwordsStDetail(object):

    def __init__(self, site_name, batch):
        assert site_name is not None, "site_name not null"
        assert batch is not None, "batch not null"

        self.site_name = site_name
        self.batch = batch
        app_name = f"{self.__class__.__name__}:{self.site_name}::{self.batch}"
        self.spark = SparkUtil.get_spark_session(app_name)
        self.hive_tb = 'dwt_merchantwords_st_detail'
        self.udf_check_utf8_and_convert_reg = F.udf(self.check_utf8_and_convert, StringType())
        pass

    @staticmethod
    def check_utf8_and_convert(word: str):
        """
        判断一个英文字符是否是utf-8编码的 即不包含乱码字符及中文 ，如果包含的话则从gbk编码转为utf-8编码
        :param str:
        :return:
        """
        import re
        # 检查是否包含有中文 如果有中文的话大概率是乱码 从gbk转为utf-8
        pattern = re.compile(r'[\u4e00-\u9fa5]')
        if pattern.match(word) is not None:
            try:
                return word.encode("gbk").decode("utf-8")
            except:
                return word
        else:
            return word

    @staticmethod
    def udf_detect_phrase_reg(lang_word_map):
        def detect_phrase(phrase: str):
            import re
            # + 号替换为空格用于分词
            phrase = re.sub(r'(\+)', ' ', phrase).strip()
            # 分词
            from nltk.tokenize import word_tokenize
            wordList = list(filter(lambda x: len(x) >= 2, word_tokenize(phrase, "english")))
            tmp_map = {
                "en": {"frequency": 0, "word": []},
                "fr": {"frequency": 0, "word": []},
                "es": {"frequency": 0, "word": []},
                "de": {"frequency": 0, "word": []},
            }
            for word in wordList:
                lang_rank_map: dict = lang_word_map.get(word)
                if lang_rank_map is not None:
                    for lang in lang_rank_map.keys():
                        frequency = lang_rank_map[lang]
                        tmp_map[lang]["frequency"] = tmp_map[lang]["frequency"] + frequency
                        tmp_map[lang]["word"].append(word)
                pass

            #  先根据word名称个数倒序后根据分数
            lang, hint_word_map = sorted(tmp_map.items(), key=lambda it: (len(it[1]['word']), it[1]['frequency']), reverse=True)[0]

            if hint_word_map['frequency'] == 0:
                return {"lang": None, "hint_word": None}
            else:
                hint_word_list = hint_word_map['word']
                hint_word = " ".join(hint_word_list)
                if len(hint_word) <= 2:
                    return {"lang": None, "hint_word": None}
                return {"lang": lang, "hint_word": hint_word}
            pass

        return F.udf(detect_phrase, MapType(StringType(), StringType()))

    def handle_calc_lang(self, df_all: DataFrame) -> DataFrame:
        lang_word_list = self.spark.sql("""
        select word, langs
        from big_data_selection.tmp_lang_word_frequency
    """).collect()
        # lang_word_df => 转为map
        lang_word_map = {row['word']: row['langs'] for row in lang_word_list}

        df_lang_all = df_all.withColumn("lang",
                                        F.coalesce(self.udf_detect_phrase_reg(lang_word_map)(F.col("keyword")).getField("lang"),
                                                   F.lit("other")))
        return df_lang_all

    def handle_update(self, date_info):
        df_append = self.spark.sql(f"""
                select keyword,
                       volume,
                       avg_3m,
                       avg_12m,
                       depth,
                       results_count,
                       sponsored_ads_count,
                       page_1_reviews,
                       appearance,
                       last_seen,
                       update_time,
                       'csv' as source_type,
                        null as api_json,
                       'us' as site_name
                from ods_merchantwords_st_detail_append
                where date_info = '{date_info}'
""").cache()
        # 判断是否需要更新
        assert df_append.count() > 0, "无需更新，请检查【ods_merchantwords_st_detail_append】数据是否异常"

        df_exist = self.spark.sql(f"""
            select keyword,
                   volume,
                   avg_3m,
                   avg_12m,
                   depth,
                   results_count,
                   sponsored_ads_count,
                   page_1_reviews,
                   appearance,
                   last_seen,
                   update_time,
                   source_type,
                   api_json,
                   site_name
            from dwt_merchantwords_st_detail;
""").cache()

        df_all = df_exist.unionByName(df_append)
        self.handle_save(df_all)

        pass

    def handle_save(self, df_all: DataFrame):
        # 处理乱码
        df_all = df_all.withColumn("keyword", F.trim(self.udf_check_utf8_and_convert_reg(F.col("keyword"))))
        #  keyword 开窗 时间倒序 取最新一个
        df_all = df_all.withColumn("row_number", F.row_number().over(window=Window.partitionBy(['keyword']).orderBy(
            F.col("update_time").desc()
        )))
        # 去重
        df_all = df_all.where("row_number == 1 ")
        df_all = df_all.drop(F.col("row_number")).cache()
        # 语种识别
        df_all = self.handle_calc_lang(df_all)
        # 分区重置
        df_all = df_all.repartition(60)

        # #  去重
        partition_dict = {
            "site_name": self.site_name,
            "batch": self.batch,
        }
        # 保存或更新
        CommonUtil.save_or_update_table(
            spark_session=self.spark,
            hive_tb_name=self.hive_tb,
            partition_dict=partition_dict,
            df_save=df_all,
            drop_exist_tmp_flag=False
        )
        print("success")
        pass

    def run(self):
        tb_ods = "tmp_merchantwords_st_detail_2024"
        if self.site_name != 'us':
            tb_ods = f"{site_name}_{tb_ods}"
            pass

        df_all = self.spark.sql(f"""
        select  
            keyword,
            volume,
            avg_3m,
            avg_12m,
            depth,
            results_count,
            sponsored_ads_count,
            page_1_reviews,
            appearance,
            last_seen,
            update_time,
            source_type,
            api_json,
            site_name
        from {tb_ods}
""")
        #  schema
        schema_api = F.schema_of_json("""
            {"historyArray": [1,2],
        "country": 1,
        "avg12Month": 1015775,
        "last_seen": 20230801,
        "totalReviewsFirstPage": 62592,
        "avg3Month": 1249957,
        "volume": 134090,
        "categoryIdArray": [
            "toys-and-games"
        ],
        "depth": 3,
        "appearance": "Evergreen",
        "popularity": 5,
        "resultsCount": 1000,
        "sponsoredCount": -1,
        "keyword": "electric scooter",
        "inLatestBatch": true
    }""")
        #         schema_csv = F.schema_of_json("""
        # {"volume":"9","depth":"8","appearance":"Rediscovered","phrase":"weaning rings for goats","sponsored ads":"0","page 1 reviews":"1914","3m avg":"3","12m avg":"1","results":"1"}
        #             """)

        df_all = df_all.withColumn("json_api", F.from_json(F.col("api_json"), schema_api))

        df_all = df_all.select(
            F.col("keyword").alias("keyword"),
            #  解析和兼容
            F.coalesce(F.col("json_api.volume"), F.col("volume")).cast(IntegerType()).alias("volume"),
            F.coalesce(F.col("json_api.avg3Month"), F.col("avg_3m")).cast(IntegerType()).alias("avg_3m"),
            F.coalesce(F.col("json_api.avg12Month"), F.col("avg_12m")).cast(IntegerType()).alias("avg_12m"),
            F.coalesce(F.col("json_api.depth"), F.col("depth")).cast(IntegerType()).alias("depth"),
            F.coalesce(F.col("json_api.resultsCount"), F.col("results_count")).cast(IntegerType()).alias("results_count"),
            F.coalesce(F.col("json_api.sponsoredCount"), F.col("sponsored_ads_count")).cast(IntegerType()).alias(
                "sponsored_ads_count"),
            F.coalesce(F.col("json_api.totalReviewsFirstPage"), F.col("page_1_reviews")).cast(IntegerType()).alias("page_1_reviews"),
            F.coalesce(F.col("json_api.appearance"), F.col("appearance")).alias("appearance"),
            F.coalesce(F.col("json_api.last_seen")).alias("last_seen"),

            F.col("update_time").alias("update_time"),
            F.col("source_type"),
            F.col("api_json"),
            F.lit(self.site_name).alias("site_name"),
            F.lit(self.batch).alias("batch")
        ).cache()
        #  保存
        self.handle_save(df_all)


if __name__ == '__main__':
    update_flag = CommonUtil.get_sys_arg(1, None)
    if update_flag == 'update':
        site_name = CommonUtil.get_sys_arg(2, 'us')
        date_info = CommonUtil.get_sys_arg(3, CommonUtil.format_now('%Y-%m-%d'))
        obj = DwtMerchantwordsStDetail('us', '2024-1').handle_update(date_info)
    else:
        site_name = CommonUtil.get_sys_arg(1, 'us')
        batch = CommonUtil.get_sys_arg(2, '2024-1')
        DwtMerchantwordsStDetail(site_name, batch).run()