etsy板块封装公共方法

Pyspark_job/dim/dim_ehunt_store_detail.py

@@ -8,7 +8,6 @@
 """
 import os
 import sys
-from itertools import chain
 
 sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
 from utils.templates import Templates
@@ -16,53 +15,7 @@ from pyspark.sql.window import Window
 from pyspark.sql import functions as F
 from utils.common_util import CommonUtil
 from utils.hdfs_utils import HdfsUtils
-
-# 国家统一为中文，key为小写去空格后的原值；未命中的保留原值
-COUNTRY_MAP = {
-    # 英文 -> 中文
-    "united states": "美国", "british": "英国", "united kingdom": "英国", "germany": "德国",
-    "canada": "加拿大", "india": "印度", "australia": "澳大利亚", "pakistan": "巴基斯坦",
-    "turkey": "土耳其", "french": "法国", "france": "法国", "ukraine": "乌克兰", "spain": "西班牙",
-    "thailand": "泰国", "indonesia": "印度尼西亚", "philippines": "菲律宾", "netherlands": "荷兰",
-    "italy": "意大利", "vietnam": "越南", "hong kong": "中国香港特区", "morocco": "摩洛哥",
-    "poland": "波兰", "china": "中国", "latvia": "拉脱维亚", "lithuania": "立陶宛", "israel": "以色列",
-    "portugal": "葡萄牙", "japan": "日本", "bulgaria": "保加利亚", "malaysia": "马来西亚",
-    "ireland": "爱尔兰", "romania": "罗马尼亚", "sweden": "瑞典", "new zealand": "新西兰",
-    "austria": "奥地利", "mexico": "墨西哥", "south korea": "韩国", "singapore": "新加坡",
-    "south africa": "南非", "switzerland": "瑞士", "argentina": "阿根廷", "hungary": "匈牙利",
-    "brazil": "巴西", "finland": "芬兰", "estonia": "爱沙尼亚", "united arab emirates": "阿联酋",
-    "belgium": "比利时", "greece": "希腊", "slovak republic": "斯洛伐克", "czech republic": "捷克",
-    "chile": "智利", "kosovo": "科索沃", "slovenia": "斯洛文尼亚", "norway": "挪威",
-    "croatia": "克罗地亚", "georgia": "格鲁吉亚", "denmark": "丹麦", "cyprus": "塞浦路斯",
-    "kazakhstan": "哈萨克斯坦", "peru": "秘鲁", "egypt": "埃及", "luxembourg": "卢森堡",
-    "nepal": "尼泊尔", "malta": "马耳他", "sri lanka": "斯里兰卡", "taiwan": "中国台湾省",
-    "armenia": "亚美尼亚", "puerto rico": "波多黎各", "ghana": "加纳", "moldova": "摩尔多瓦",
-    "u.s. virgin islands": "美属维尔京群岛", "saudi arabia": "沙特阿拉伯", "nigeria": "尼日利亚",
-    "kenya": "肯尼亚", "venezuela": "委内瑞拉", "colombia": "哥伦比亚", "russia": "俄罗斯",
-    "costa rica": "哥斯达黎加", "guatemala": "危地马拉", "iceland": "冰岛",
-    "dominican republic": "多米尼加共和国", "ecuador": "厄瓜多尔", "lebanon": "黎巴嫩",
-    "vzbekistan": "乌兹别克斯坦", "uzbekistan": "乌兹别克斯坦", "albania": "阿尔巴尼亚",
-    "jordan": "约旦", "belarus": "白俄罗斯", "monaco": "摩纳哥", "tanzania": "坦桑尼亚",
-    "kyrgyzstan": "吉尔吉斯斯坦", "uruguay": "乌拉圭", "algeria": "阿尔及利亚", "antarctica": "南极洲",
-    "belize": "伯利兹", "ethiopia": "埃塞俄比亚", "haiti": "海地", "kuwait": "科威特",
-    "macedonia": "北马其顿", "nicaragua": "尼加拉瓜", "panama": "巴拿马", "tunisia": "突尼斯",
-    "niue": "纽埃", "bolivia": "玻利维亚", "montenegro": "黑山", "barbados": "巴巴多斯",
-    "bahrain": "巴林", "honduras": "洪都拉斯", "azerbaijan": "阿塞拜疆", "turkmenistan": "土库曼斯坦",
-    "qatar": "卡塔尔", "afghanistan": "阿富汗", "macao": "中国澳门特区", "macau": "中国澳门特区",
-    "oman": "阿曼", "jamaica": "牙买加", "guyana": "圭亚那", "guam": "关岛", "faroe islands": "法罗群岛",
-    "cayman islands": "开曼群岛", "paraguay": "巴拉圭", "bosnia and herzegovina": "波黑",
-    "bangladesh": "孟加拉", "el salvador": "萨尔瓦多", "cameroon": "喀麦隆", "botswana": "博茨瓦纳",
-    "bermuda": "百慕大", "bahamas": "巴哈马", "andorra": "安道尔", "zambia": "赞比亚",
-    "uganda": "乌干达", "togo": "多哥", "sudan": "苏丹", "senegal": "塞内加尔", "samoa": "萨摩亚",
-    "palestinian territory": "巴勒斯坦", "netherlands antilles": "荷属安的列斯", "namibia": "纳米比亚",
-    "myanmar": "缅甸", "mongolia": "蒙古", "marshall islands": "马绍尔群岛", "maldives": "马尔代夫",
-    "liechtenstein": "列支敦士登", "laos": "老挝", "greenland": "格陵兰",
-    # 中文非标准写法 -> 标准写法
-    "拉托维亚": "拉脱维亚", "斯罗文尼亚": "斯洛文尼亚", "塞普路斯": "塞浦路斯",
-    "科索沃,塞尔维亚": "科索沃", "法属尼留旺岛": "留尼汪",
-}
-YES_VALUES = ['yes', 'y', 'true', '1', '是']
-NO_VALUES = ['no', 'n', 'false', '0', '否']
+from utils.ehunt_util import country_to_cn, yes_no_to_cn
 
 
 class DimEhuntStoreDetail(Templates):
@@ -99,11 +52,6 @@ class DimEhuntStoreDetail(Templates):
         self.df_store = self.spark.sql(sqlQuery=sql).cache()
         print("ods数据量为：", self.df_store.count())
 
-    @staticmethod
-    def handle_yes_no(col_name):
-        val = F.lower(F.trim(F.col(col_name)))
-        return F.when(val.isin(YES_VALUES), F.lit("是")).when(val.isin(NO_VALUES), F.lit("否")).otherwise(F.lit(None))
-
     # store_name去重，按入库时间取最新一条
     def handle_store_distinct(self):
         window = Window.partitionBy('store_name').orderBy(F.col('created_at').desc_nulls_last())
@@ -128,21 +76,14 @@ class DimEhuntStoreDetail(Templates):
     # 明星卖家/Raving标签统一为 是/否
     def handle_flag(self):
         self.df_store = self.df_store.withColumn(
-            'is_star', self.handle_yes_no('is_star')
+            'is_star', yes_no_to_cn('is_star')
         ).withColumn(
-            'is_raving', self.handle_yes_no('is_raving')
+            'is_raving', yes_no_to_cn('is_raving')
         )
 
     # 国家统一为中文
     def handle_country(self):
-        country_map = F.create_map([F.lit(x) for x in chain(*COUNTRY_MAP.items())])
-        country_key = F.lower(F.trim(F.col('country')))
-        self.df_store = self.df_store.withColumn(
-            'country',
-            F.when(
-                country_key.isNull() | country_key.isin('', '[]'), F.lit(None)
-            ).otherwise(F.coalesce(country_map[country_key], F.trim(F.col('country'))))
-        )
+        self.df_store = self.df_store.withColumn('country', country_to_cn('country'))
 
     def handle_column(self):
         self.df_save = self.df_store.select(

Pyspark_job/utils/ehunt_util.py

+"""
+   @Author      : CT
+   @Description : Etsy(eHunt)店铺/产品共用清洗规则：国家统一中文、是/否标签统一
+"""
+from itertools import chain
+
+from pyspark.sql import functions as F
+
+# 国家统一为中文，key为小写去空格后的原值；未命中的保留原值
+COUNTRY_MAP = {
+    # 英文 -> 中文
+    "united states": "美国", "british": "英国", "united kingdom": "英国", "germany": "德国",
+    "canada": "加拿大", "india": "印度", "australia": "澳大利亚", "pakistan": "巴基斯坦",
+    "turkey": "土耳其", "french": "法国", "france": "法国", "ukraine": "乌克兰", "spain": "西班牙",
+    "thailand": "泰国", "indonesia": "印度尼西亚", "philippines": "菲律宾", "netherlands": "荷兰",
+    "italy": "意大利", "vietnam": "越南", "hong kong": "中国香港特区", "morocco": "摩洛哥",
+    "poland": "波兰", "china": "中国", "latvia": "拉脱维亚", "lithuania": "立陶宛", "israel": "以色列",
+    "portugal": "葡萄牙", "japan": "日本", "bulgaria": "保加利亚", "malaysia": "马来西亚",
+    "ireland": "爱尔兰", "romania": "罗马尼亚", "sweden": "瑞典", "new zealand": "新西兰",
+    "austria": "奥地利", "mexico": "墨西哥", "south korea": "韩国", "singapore": "新加坡",
+    "south africa": "南非", "switzerland": "瑞士", "argentina": "阿根廷", "hungary": "匈牙利",
+    "brazil": "巴西", "finland": "芬兰", "estonia": "爱沙尼亚", "united arab emirates": "阿联酋",
+    "belgium": "比利时", "greece": "希腊", "slovak republic": "斯洛伐克", "czech republic": "捷克",
+    "chile": "智利", "kosovo": "科索沃", "slovenia": "斯洛文尼亚", "norway": "挪威",
+    "croatia": "克罗地亚", "georgia": "格鲁吉亚", "denmark": "丹麦", "cyprus": "塞浦路斯",
+    "kazakhstan": "哈萨克斯坦", "peru": "秘鲁", "egypt": "埃及", "luxembourg": "卢森堡",
+    "nepal": "尼泊尔", "malta": "马耳他", "sri lanka": "斯里兰卡", "taiwan": "中国台湾省",
+    "armenia": "亚美尼亚", "puerto rico": "波多黎各", "ghana": "加纳", "moldova": "摩尔多瓦",
+    "u.s. virgin islands": "美属维尔京群岛", "saudi arabia": "沙特阿拉伯", "nigeria": "尼日利亚",
+    "kenya": "肯尼亚", "venezuela": "委内瑞拉", "colombia": "哥伦比亚", "russia": "俄罗斯",
+    "costa rica": "哥斯达黎加", "guatemala": "危地马拉", "iceland": "冰岛",
+    "dominican republic": "多米尼加共和国", "ecuador": "厄瓜多尔", "lebanon": "黎巴嫩",
+    "vzbekistan": "乌兹别克斯坦", "uzbekistan": "乌兹别克斯坦", "albania": "阿尔巴尼亚",
+    "jordan": "约旦", "belarus": "白俄罗斯", "monaco": "摩纳哥", "tanzania": "坦桑尼亚",
+    "kyrgyzstan": "吉尔吉斯斯坦", "uruguay": "乌拉圭", "algeria": "阿尔及利亚", "antarctica": "南极洲",
+    "belize": "伯利兹", "ethiopia": "埃塞俄比亚", "haiti": "海地", "kuwait": "科威特",
+    "macedonia": "北马其顿", "nicaragua": "尼加拉瓜", "panama": "巴拿马", "tunisia": "突尼斯",
+    "niue": "纽埃", "bolivia": "玻利维亚", "montenegro": "黑山", "barbados": "巴巴多斯",
+    "bahrain": "巴林", "honduras": "洪都拉斯", "azerbaijan": "阿塞拜疆", "turkmenistan": "土库曼斯坦",
+    "qatar": "卡塔尔", "afghanistan": "阿富汗", "macao": "中国澳门特区", "macau": "中国澳门特区",
+    "oman": "阿曼", "jamaica": "牙买加", "guyana": "圭亚那", "guam": "关岛", "faroe islands": "法罗群岛",
+    "cayman islands": "开曼群岛", "paraguay": "巴拉圭", "bosnia and herzegovina": "波黑",
+    "bangladesh": "孟加拉", "el salvador": "萨尔瓦多", "cameroon": "喀麦隆", "botswana": "博茨瓦纳",
+    "bermuda": "百慕大", "bahamas": "巴哈马", "andorra": "安道尔", "zambia": "赞比亚",
+    "uganda": "乌干达", "togo": "多哥", "sudan": "苏丹", "senegal": "塞内加尔", "samoa": "萨摩亚",
+    "palestinian territory": "巴勒斯坦", "netherlands antilles": "荷属安的列斯", "namibia": "纳米比亚",
+    "myanmar": "缅甸", "mongolia": "蒙古", "marshall islands": "马绍尔群岛", "maldives": "马尔代夫",
+    "liechtenstein": "列支敦士登", "laos": "老挝", "greenland": "格陵兰", "cambodia": "柬埔寨",
+    # 中文非标准写法 -> 标准写法
+    "拉托维亚": "拉脱维亚", "斯罗文尼亚": "斯洛文尼亚", "塞普路斯": "塞浦路斯",
+    "科索沃,塞尔维亚": "科索沃", "法属尼留旺岛": "留尼汪", "纽爱": "纽埃",
+}
+YES_VALUES = ['yes', 'y', 'true', '1', '是']
+NO_VALUES = ['no', 'n', 'false', '0', '否']
+
+
+def country_to_cn(col_name):
+    """国家统一为中文，空串/[] 置空，字典未命中保留原值"""
+    country_map = F.create_map([F.lit(x) for x in chain(*COUNTRY_MAP.items())])
+    country_key = F.lower(F.trim(F.col(col_name)))
+    return F.when(
+        country_key.isNull() | country_key.isin('', '[]'), F.lit(None)
+    ).otherwise(F.coalesce(country_map[country_key], F.trim(F.col(col_name))))
+
+
+def yes_no_to_cn(col_name):
+    """标签统一为 是/否，无法识别的置空"""
+    val = F.lower(F.trim(F.col(col_name)))
+    return F.when(val.isin(YES_VALUES), F.lit("是")).when(val.isin(NO_VALUES), F.lit("否")).otherwise(F.lit(None))