Skip to content

A
Amazon-Selection-Data
Commit 57100823 by chenyuanjie
Options

no message

parent b8e21eea
Showing with 8 additions and 1998 deletions
.gitignore
......@@ -2,3 +2,5 @@
.venv/
.gitignore
Pyspark_job/utils/__pycache__/
Pyspark_job/ct
Pyspark_job/sqoop_ct
.idea/Amazon-Selection-Data.iml
<?xml version="1.0" encoding="UTF-8"?>
<module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$" />
<orderEntry type="inheritedJdk" />
<content url="file://$MODULE_DIR$">
<sourceFolder url="file://$MODULE_DIR$/Pyspark_job" isTestSource="false" />
</content>
<orderEntry type="jdk" jdkName="Python 3.8 (Amazon-Selection)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
<component name="PyDocumentationSettings">
......
.idea/misc.xml
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.9" project-jdk-type="Python SDK" />
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.8 (Amazon-Selection)" project-jdk-type="Python SDK" />
</project>
\ No newline at end of file
Pyspark_job/ct/aba_2023_10_12_export.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.spark_util import SparkUtil
from pyspark.sql.functions import col
if __name__ == '__main__':
spark = SparkUtil.get_spark_session("ABA_2023_10_12_export")
sql1 = """
select
date_info,
search_term,
st_bsr_cate_1_id_new as category_id,
market_cycle_type,
is_first_text,
is_ascending_text,
is_high_return_text,
is_search_text,
st_movie_label,
st_brand_label,
bsr_orders,
st_word_num,
st_num,
rank
from dwt_aba_st_analytics
where site_name = 'us'
and date_type = 'month'
and date_info in ('2023-10','2023-11','2023-12');
"""
df_dwt_aba_st_analytics = spark.sql(sql1).cache()
sql2 = """
select
category_id,
en_name
from dim_bsr_category_tree
where site_name = 'us'
and category_parent_id = 0;
"""
df_dim_bsr_category_tree = spark.sql(sql2).cache()
sql3 = """
select
search_term,
rank_change_rate,
rank_rate_of_change,
date_info
from dwt_aba_last_change_rate
where site_name = 'us'
and date_type = 'month'
and date_info in ('2023-10','2023-11','2023-12');
"""
df_dwt_aba_last_change_rate = spark.sql(sql3).cache()
# 过滤出满足条件的词
df_dwt_aba_st_analytics = df_dwt_aba_st_analytics.filter(
"(is_first_text = 1) or (is_ascending_text = 1) or (market_cycle_type in (1, 2))"
)
df_save = df_dwt_aba_st_analytics.join(
df_dim_bsr_category_tree, on='category_id', how='left'
).join(
df_dwt_aba_last_change_rate, on=['date_info', 'search_term'], how='left'
)
df_save = df_save.select(
col('date_info').alias('year_month'),
col('search_term'),
col('en_name').alias('category'),
col('market_cycle_type'),
col('is_first_text'),
col('is_ascending_text'),
col('is_high_return_text'),
col('is_search_text'),
col('st_movie_label').alias('movie_label'),
col('st_brand_label').alias('brand_label'),
col('bsr_orders'),
col('st_word_num').alias('word_counts'),
col('st_num').alias('word_frequency'),
col('rank'),
col('rank_change_rate').alias('year_on_year'),
col('rank_rate_of_change').alias('month_on_month')
)
df_save.repartition(5).show(10, truncate=True)
df_save.write.saveAsTable(name='tmp_aba_2023_export', format='hive', mode='append')
spark.stop()
Pyspark_job/ct/aba_2023_10_12_word_frequency.py deleted 100644 → 0
import os
import re
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.functions import count, explode, split, udf, lit
from pyspark.sql.types import ArrayType, StringType
if __name__ == '__main__':
date_info = CommonUtil.get_sys_arg(1, None)
spark = SparkUtil.get_spark_session("ABA_2023_10_12_word_frequency")
# 自定义函数，将词组拆分为2个单词为一组
def split_tow_by_tow(search_term):
words = search_term.split()
pairs = []
for i in range(len(words) - 1):
pairs.append(words[i] + ' ' + words[i + 1])
return pairs
u_split_tow_by_tow = udf(split_tow_by_tow, ArrayType(StringType()))
# 自定义函数，将词组拆分为3个单词为一组
def split_three_by_three(search_term):
words = search_term.split()
triplets = []
for i in range(len(words) - 2):
triplets.append(words[i] + ' ' + words[i + 1] + ' ' + words[i + 2])
return triplets
u_split_three_by_three = udf(split_three_by_three, ArrayType(StringType()))
# 自定义函数，剔除掉多余字符
def characters_to_remove(search_term):
pattern = r'\s[^\w\s%\']+?\s'
cleaned_text = re.sub(pattern, ' ', search_term)
cleaned_text = cleaned_text.replace('\n', ' ')
return cleaned_text
u_characters_to_remove = udf(characters_to_remove, StringType())
sql = f"""
select
search_term
from dwt_aba_st_analytics
where site_name = 'us'
and date_type = 'month'
and date_info = '{date_info}';
"""
df_aba = spark.sql(sql).cache()
df_aba = df_aba.select(
u_characters_to_remove(df_aba['search_term']).alias('search_term')
)
df_one_word = df_aba.select(
explode(split(df_aba['search_term'], ' ')).alias('word')
).groupby(
['word']
).agg(
count('word').alias('word_frequency')
).filter(
'word_frequency >= 50'
).withColumn(
'date_info',
lit(f'{date_info}-1')
)
df_tow_word = df_aba.select(
explode(u_split_tow_by_tow(df_aba['search_term'])).alias('word')
).groupby(
['word']
).agg(
count('word').alias('word_frequency')
).filter(
'word_frequency >= 50'
).withColumn(
'date_info',
lit(f'{date_info}-2')
)
df_three_word = df_aba.select(
explode(u_split_three_by_three(df_aba['search_term'])).alias('word')
).groupby(
['word']
).agg(
count('word').alias('word_frequency')
).filter(
'word_frequency >= 50'
).withColumn(
'date_info',
lit(f'{date_info}-3')
)
df_one_word.write.saveAsTable(name='tmp_word_frequency', format='hive', mode='append', partitionBy='date_info')
df_tow_word.write.saveAsTable(name='tmp_word_frequency', format='hive', mode='append', partitionBy='date_info')
df_three_word.write.saveAsTable(name='tmp_word_frequency', format='hive', mode='append', partitionBy='date_info')
Pyspark_job/ct/aba_2023_year_word_frequency.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.spark_util import SparkUtil
from pyspark.sql.functions import count, explode, lit, desc, sum
from pyspark.sql.types import ArrayType, StringType
from textblob import Word
from googletrans import Translator
class ABA2023YearWordFrequency(object):
def __init__(self):
self.spark = SparkUtil.get_spark_session("spark_task: aba_2023_year_word_frequency")
self.df_aba_2023 = self.spark.sql(f"select 1+1;")
self.df_beside_category = self.spark.sql(f"select 1+1;")
self.df_translate = self.spark.sql(f"select 1+1;")
self.df_save = self.spark.sql(f"select 1+1;")
self.df_save1 = self.spark.sql(f"select 1+1;")
self.df_save2 = self.spark.sql(f"select 1+1;")
self.df_agg = self.spark.sql(f"select 1+1;")
# 自定义udf
self.u_get_singular_form = self.spark.udf.register('get_singular_form', self.get_singular_form, StringType())
self.u_word_tokenize = self.spark.udf.register('word_tokenize', self.word_tokenize, ArrayType(StringType()))
# self.u_word_translate = self.spark.udf.register('word_translate', self.word_translate, StringType())
@staticmethod
def get_singular_form(word: str):
"""
将单词全部转化为单数形式
"""
if word:
singular_form = Word(word).lemmatize("n")
# word_object = Word(word)
# singular_form = word_object.singularize()
return singular_form
return word
@staticmethod
def word_tokenize(title: str):
"""
分词器
"""
from nltk.tokenize import word_tokenize
result = word_tokenize(title, "english")
return result
# @staticmethod
# def word_translate(word: str):
# if word:
# try:
# translator = Translator()
# result = translator.translate(word, src='en', dest='zh-cn')
# return result.text
# except Exception as e:
# # 处理其他未知错误
# print(f"An unexpected error occurred: {e}")
# return None
# return None
def read_data(self):
sql1 = f"""
select
search_term,
category_id
from dwt_aba_last365
where site_name = 'us'
and date_type = 'last365day'
and date_info = '2023-12';
"""
self.df_aba_2023 = self.spark.sql(sql1).cache()
print("df_aba_2023的数量：")
print(self.df_aba_2023.count())
sql2 = f"""
select
category_id
from dim_bsr_category_tree
where site_name = 'us'
and en_name in ('Audible Books & Originals', 'Books', 'Kindle Store', 'Apps & Games', 'Movies & TV', 'CDs & Vinyl', 'Software', 'Video Games')
and category_parent_id = 0;
"""
self.df_beside_category = self.spark.sql(sql2).cache()
print("df_beside_category的数量：")
print(self.df_beside_category.count())
sql3 = f"""
select
word,
simple_cn as cn
from tmp_en_dict;
"""
self.df_translate = self.spark.sql(sql3).cache()
print("df_translate的数量：")
print(self.df_translate.count())
def handle_data(self):
self.df_save = self.df_aba_2023.join(
self.df_beside_category, on='category_id', how='left_anti'
).select('search_term')
self.df_save = self.df_save.select(explode(self.u_word_tokenize(self.df_save['search_term'])).alias('word'))
self.df_save = self.df_save.groupby(['word']).agg(
count('word').alias('word_frequency')
)
self.df_save = self.df_save.join(
self.df_translate, on='word', how='left'
).withColumn(
'word_singular_form',
self.u_get_singular_form(self.df_save['word'])
).cache()
self.df_save1 = self.df_save.select(
'word', 'word_frequency', 'cn'
).orderBy(
desc('word_frequency')
).withColumn(
'date_info',
lit('2023')
)
print("df_save1的数量：")
print(self.df_save1.count())
self.df_save1.write.saveAsTable(name='tmp_word_frequency', format='hive', mode='append', partitionBy='date_info')
print("df_save1存储完成！")
self.df_agg = self.df_save.groupby(['word_singular_form']).agg(
sum('word_frequency').alias('word_frequency')
)
self.df_save2 = self.df_save.select('word', 'cn', 'word_singular_form').join(
self.df_agg, on='word_singular_form', how='left'
).select(
'word', 'word_frequency', 'cn'
).orderBy(
desc('word_frequency')
).withColumn(
'date_info',
lit('2023-merge')
)
print("df_save2的数量：")
print(self.df_save2.count())
self.df_save2.write.saveAsTable(name='tmp_word_frequency', format='hive', mode='append', partitionBy='date_info')
print("df_save2存储完成！")
if __name__ == '__main__':
obj = ABA2023YearWordFrequency()
obj.read_data()
obj.handle_data()
Pyspark_job/ct/aba_re_run_month.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.DolphinschedulerHelper import DolphinschedulerHelper
from utils.common_util import CommonUtil
from utils.spark_util import SparkUtil
if __name__ == '__main__':
start_date = CommonUtil.get_sys_arg(1, None)
end_date = CommonUtil.get_sys_arg(2, None)
spark_session = SparkUtil.get_spark_session("re-run-aba-month")
sql = f"""
select distinct year_month as date_info from dim_date_20_to_30 where year_month >= '{start_date}' and year_month < '{end_date}';
"""
date_df = spark_session.sql(sql)
print(date_df.show())
date_list = sorted([d.asDict().get("date_info") for d in date_df.collect()])
print(date_list)
for date_info in date_list:
startParams = {
"site_name": "us",
"date_type": "month",
"date_info": date_info
}
print(startParams)
DolphinschedulerHelper.start_and_watch_process_instance(
"big_data_selection",
process_df_name='月-重跑ABA四分位',
startParams={
"site_name": "us",
"date_type": "month",
"date_info": date_info
}
)
CommonUtil.send_wx_msg(["huangjian", "chenyuanjie"], "【月-重跑ABA四分位】重跑完成", "")
pass
Pyspark_job/ct/asin_to_number.py deleted 100644 → 0
def asin_to_number(asin):
"""
Convert a 10-character ASIN string to a unique number.
This function assumes that ASIN consists of uppercase letters and digits.
"""
def char_to_number(char):
if char.isdigit():
return int(char)
else:
return ord(char) - 55 # 'A' -> 10, 'B' -> 11, ..., 'Z' -> 35
if len(asin) != 10:
raise ValueError("ASIN must be 10 characters long")
base = 36
asin_number = 0
for i, char in enumerate(reversed(asin)):
asin_number += char_to_number(char) * (base ** i)
# The final number is taken modulo 1 billion to fit the range 1-10 billion
return asin_number % 1000000000
if __name__ == '__main__':
x = asin_to_number('B0CGY4LZQ3')
print(x)
s = f'us_asin_image_part{int(x / 1000_0000) + 1}'
print(s)
Pyspark_job/ct/dwt_merchantwords_st_detail_merge.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from utils.templates import Templates
from utils.hdfs_utils import HdfsUtils
from utils.spark_util import SparkUtil
from pyspark.sql.window import Window
from pyspark.storagelevel import StorageLevel
from pyspark.sql import functions as F
class DwtMerchantwordsStDetailMerge(Templates):
def __init__(self, site_name='us'):
super().__init__()
self.site_name = site_name
self.batch = '2024-1'
self.db_save = 'dwt_merchantwords_st_detail_merge'
self.spark = self.create_spark_object(
app_name=f"DwtMerchantwordsStDetailMerge: {self.site_name}, {self.batch}")
self.partitions_num = 15
self.partitions_by = ['site_name', 'batch']
self.df = self.spark.sql(f"select 1+1;")
self.df_save = self.spark.sql(f"select 1+1;")
hdfs_path = f"/home/{SparkUtil.DEF_USE_DB}/dwt/{self.db_save}/site_name={self.site_name}/batch={self.batch}"
print(f"清除hdfs目录中.....{hdfs_path}")
HdfsUtils.delete_hdfs_file(hdfs_path)
def read_data(self):
print("读取dwt_merchantwords_st_detail数据")
sql = f"""
select
keyword,
volume,
avg_3m,
avg_12m,
depth,
results_count,
sponsored_ads_count,
page_1_reviews,
appearance,
last_seen,
update_time,
lang,
batch as last_batch
from dwt_merchantwords_st_detail
where site_name = '{self.site_name}'
and batch in ('2023-1', '2024-1');
"""
self.df = self.spark.sql(sqlQuery=sql)
self.df = self.df.repartition(80).persist(StorageLevel.MEMORY_ONLY)
self.df.show(10, truncate=True)
def handle_data(self):
window = Window.partitionBy('keyword').orderBy(
F.desc_nulls_last('last_batch')
)
self.df = self.df.withColumn("u_rank", F.row_number().over(window=window))
self.df = self.df.filter('u_rank=1').drop('u_rank')
self.df_save = self.df.withColumn(
'site_name',
F.lit(self.site_name)
).withColumn(
'batch',
F.lit(self.batch)
)
if __name__ == '__main__':
site_name = sys.argv[1]
handle_obj = DwtMerchantwordsStDetailMerge(site_name=site_name)
handle_obj.run()
Pyspark_job/ct/es_to_hive.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.hdfs_utils import HdfsUtils
from utils.common_util import CommonUtil
from utils.templates import Templates
from pyspark.sql import functions as F
class FlowAsinLast30days(Templates):
def __init__(self):
super().__init__()
self.db_save = "tmp_flow_asin_last30days"
self.spark = self.create_spark_object(app_name="FlowAsinLast30days")
self.partitions_num = 20
self.partition_dict = {}
self.df_es = self.spark.sql(f"select 1+1;")
self.df_parent = self.spark.sql(f"select 1+1;")
self.df_joined = self.spark.sql(f"select 1+1;")
self.df_save = self.spark.sql(f"select 1+1;")
def read_data(self):
self.df_es = self.spark.read.format("org.elasticsearch.spark.sql")\
.option("es.nodes", "192.168.10.217")\
.option("es.port", "9200")\
.option("es.net.http.auth.user", "elastic")\
.option("es.net.http.auth.pass", "selection2021.+")\
.option("es.resource", "us_st_detail_last_4_week")\
.option("es.query", '{"query": {"match_all": {}}}')\
.load()
columns = ["asin", "first_category_rank", "asin_bought_month", "total_comments", "variation_num", "site_name", "account_name"]
self.df_es = self.df_es.select(columns).cache()
self.df_es.show()
sql = f"""
select
asin,
parent_asin
from
ods_asin_variat;
"""
self.df_parent = self.spark.sql(sqlQuery=sql).cache()
def handle_data(self):
# self.df_parent = self.df_parent.groupby(["parent_asin"]).agg(F.count("asin").alias("variation_num"))
self.df_joined = self.df_es.join(self.df_parent, "asin", "left")
self.df_joined = self.df_joined\
.withColumn("parent_asin_is_null", F.when(F.col("parent_asin").isNull(), F.lit(1)).otherwise(F.lit(0)))\
.withColumn("parent_asin_exist", F.when(F.col("parent_asin").isNotNull(), F.lit(1)).otherwise(F.lit(0)))
def save_data(self):
self.df_save = self.df_joined
hdfs_path_asin_info = CommonUtil.build_hdfs_path(self.db_save, partition_dict=self.partition_dict)
print(f"清除hdfs目录中:{hdfs_path_asin_info}")
HdfsUtils.delete_file_in_folder(hdfs_path_asin_info)
print(f"当前存储的表名为：{self.db_save}")
self.df_save.write.saveAsTable(name=self.db_save, format='hive', mode='append')
print("success")
if __name__ == '__main__':
obj = FlowAsinLast30days()
obj.run()
\ No newline at end of file
Pyspark_job/ct/export_dwt_flow_asin.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.DolphinschedulerHelper import DolphinschedulerHelper
from utils.common_util import CommonUtil
if __name__ == '__main__':
date_list = ["2024-02","2024-01","2023-12","2023-11","2023-10","2023-09"]
for date_info in date_list:
startParams = {
"site_name": "us",
"date_type": "month",
"date_info": date_info
}
print(startParams)
DolphinschedulerHelper.start_and_watch_process_instance(
"big_data_selection",
process_df_name='export_dwt_flow_asin_api',
startParams=startParams
)
CommonUtil.send_wx_msg(["chenyuanjie", "wangrui4"], "【export_dwt_flow_asin_api】导出完成", "")
pass
Pyspark_job/ct/merchantwords_hive_to_pg16.py deleted 100644 → 0
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.functions import row_number, lit
from pyspark.sql.window import Window
from pyspark.sql.types import StringType, ArrayType
from urllib.parse import quote
from datetime import datetime
if __name__ == '__main__':
date_info = CommonUtil.get_sys_arg(1, None)
n = CommonUtil.get_sys_arg(2, 0)
hive_tb = "dwt_merchantwords_st_detail"
export_tb = "us_merchantwords_search_term_month_syn_2024"
spark = SparkUtil.get_spark_session(f"export: {hive_tb}")
# 一次导出400w条数据
batch_size = (int(n)-1) * 4000000
start_index = 1 + batch_size
end_index = 4000000 + batch_size
# 构建 URL 的函数
def build_urls(search_term):
url_template = f"https://www.amazon.com/s?k={{search_term}}&page={{page_number}}"
search_term_chinese = quote(search_term, 'utf-8')
search_term_chinese = search_term_chinese.replace("'", '%27').replace("/", '%2F')
urls = [
url_template.format(
search_term=search_term_chinese.replace(' ', '+').replace('&', '%26').replace('#', '%23').replace('(',
'%28').replace(
')', '%29'), page_number=1),
url_template.format(
search_term=search_term_chinese.replace(' ', '+').replace('&', '%26').replace('#', '%23').replace('(',
'%28').replace(
')', '%29'), page_number=2),
url_template.format(
search_term=search_term_chinese.replace(' ', '+').replace('&', '%26').replace('#', '%23').replace('(',
'%28').replace(
')', '%29'), page_number=3)
]
return urls
# 将Python函数转换为UDF
spark.udf.register("build_urls", build_urls, ArrayType(StringType()))
# 从 PostgreSQL 数据库中读取已有数据
# df_pg = spark.read.format("jdbc") \
# .option("url", "jdbc:postgresql://192.168.10.225:5432/selection") \
# .option("dbtable", export_tb) \
# .option("user", "yswg_postgres") \
# .option("password", "yswg_postgres") \
# .load()
# df_pg = df_pg\
# .select("search_term") \
# .drop_duplicates(["search_term"]) \
# .repartition(70) \
# .cache()
# 从 Hive 表中读取数据
df_hive = spark.sql(f"SELECT keyword FROM {hive_tb}")
df_hive = df_hive\
.withColumn("row_num", row_number().over(Window.orderBy("keyword")))\
.filter(f"row_num BETWEEN {start_index} AND {end_index}")\
.select("keyword")\
.repartition(10) \
.cache()
# 过滤掉keyword含有中文的数据
df_hive = df_hive.filter(~df_hive["keyword"].rlike("[\u4e00-\u9fff]"))
# 过滤掉已存在于目标数据库中的数据
# df_hive = df_hive.join(df_pg, df_hive["keyword"] == df_pg["search_term"], "leftanti")
# 如果没有数据需要导出，退出循环
if df_hive.count() == 0:
print("-------数据已全部导出！-------")
quit()
df_hive = df_hive.selectExpr("keyword AS search_term")
df_hive = df_hive.selectExpr("search_term", "explode(build_urls(search_term)) AS url")
df_hive = df_hive.withColumn("date_info", lit(date_info))
# 导出数据到 PostgreSQL 数据库
df_hive.write.format("jdbc") \
.option("url", "jdbc:postgresql://192.168.10.225:5432/selection") \
.option("dbtable", export_tb) \
.option("user", "yswg_postgres") \
.option("password", "yswg_postgres") \
.mode("append") \
.save()
spark.stop()
\ No newline at end of file
Pyspark_job/ct/merchantwords_sr_to_pg16.py deleted 100644 → 0
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.functions import row_number, lit, length
from pyspark.sql.window import Window
from pyspark.sql.types import StringType, ArrayType
from urllib.parse import quote
if __name__ == '__main__':
date_info = CommonUtil.get_sys_arg(1, None)
n = CommonUtil.get_sys_arg(2, 0)
import_tb = "search_term_result_year"
export_tb = "us_merchantwords_search_term_month_syn_2024"
spark = SparkUtil.get_spark_session("MerchantwordsSRToPG16")
# 一次导出400w条数据
batch_size = (int(n)-1) * 4000000
start_index = 1 + batch_size
end_index = 4000000 + batch_size
# 构建 URL 的函数
def build_urls(search_term):
url_template = f"https://www.amazon.com/s?k={{search_term}}&page={{page_number}}"
search_term_chinese = quote(search_term, 'utf-8')
search_term_chinese = search_term_chinese.replace("'", '%27').replace("/", '%2F')
urls = [
url_template.format(
search_term=search_term_chinese.replace(' ', '+').replace('&', '%26').replace('#', '%23').replace('(',
'%28').replace(
')', '%29'), page_number=1),
url_template.format(
search_term=search_term_chinese.replace(' ', '+').replace('&', '%26').replace('#', '%23').replace('(',
'%28').replace(
')', '%29'), page_number=2),
url_template.format(
search_term=search_term_chinese.replace(' ', '+').replace('&', '%26').replace('#', '%23').replace('(',
'%28').replace(
')', '%29'), page_number=3)
]
return urls
# 将Python函数转换为UDF
spark.udf.register("build_urls", build_urls, ArrayType(StringType()))
# 从SR数据库中读取已有数据
df = spark.read.format("jdbc") \
.option("url", "jdbc:mysql://192.168.10.151:19030/test") \
.option("dbtable", import_tb) \
.option("user", "chenyuanjie") \
.option("password", "chenyuanjie12345") \
.load()
df = df.withColumn(
"row_num",
row_number().over(Window.orderBy("search_term"))
).filter(f"row_num BETWEEN {start_index} AND {end_index}").repartition(20).cache()
# 过滤掉keyword含有中文的数据
df = df.filter(~df["search_term"].rlike("[\u4e00-\u9fff]"))
# 如果没有数据需要导出，退出循环
if df.count() == 0:
print("-------数据已全部导出！-------")
quit()
df = df.selectExpr("search_term", "explode(build_urls(search_term)) AS url")
df = df.filter(length(df['url']) <= 450)
df = df.withColumn("date_info", lit(date_info))
# 导出数据到 PostgreSQL 数据库
df.write.format("jdbc") \
.option("url", "jdbc:postgresql://192.168.10.225:5432/selection") \
.option("dbtable", export_tb) \
.option("user", "yswg_postgres") \
.option("password", "yswg_postgres") \
.mode("append") \
.save()
spark.stop()
\ No newline at end of file
Pyspark_job/ct/ods_asin_detail_sr_to_hive.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.spark_util import SparkUtil
from utils.StarRocksHelper import StarRocksHelper
from utils.common_util import CommonUtil
from utils.hdfs_utils import HdfsUtils
if __name__ == '__main__':
spark = SparkUtil.get_spark_session("ods_asin_detail_sr_to_hive")
partition_dict = {
"site_name": 'us',
"date_type": 'month',
"date_info": '2024-03'
}
hdfs_path = CommonUtil.build_hdfs_path('ods_asin_detail_test', partition_dict=partition_dict)
HdfsUtils.delete_hdfs_file(hdfs_path)
connection_info = StarRocksHelper.get_connection_info('selection')
df_sr = spark.read.format("starrocks") \
.option("starrocks.fe.http.url", f"{connection_info['ip']}:{connection_info['http_port']}") \
.option("starrocks.fe.jdbc.url", f"jdbc:mysql://{connection_info['ip']}:{connection_info['jdbc_port']}") \
.option("starrocks.table.identifier", "test.ods_asin_detail_test2") \
.option("starrocks.user", connection_info['user']) \
.option("starrocks.password", connection_info['pwd']) \
.option("starrocks.request.tablet.size", "1") \
.option("starrocks.batch.size", "40960") \
.option("starrocks.exec.mem.limit", "21474836480") \
.load()
print("读取完毕")
df_sr.repartition(50)
partitions_by = ['site_name', 'date_type', 'date_info']
df_sr.write.saveAsTable(name='ods_asin_detail_test', format='hive', mode='append', partitionBy=partitions_by)
spark.stop()
# 创建lzo索引和修复元数据
CommonUtil.after_import(hdfs_path=hdfs_path, hive_tb='ods_asin_detail_test')
Pyspark_job/ct/ods_asin_detail_to_sr.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.spark_util import SparkUtil
from utils.StarRocksHelper import StarRocksHelper
if __name__ == '__main__':
spark = SparkUtil.get_spark_session("ods_asin_detail_to_sr_test")
sql = """
select
*
from ods_asin_detail
where site_name = 'us'
and date_type = 'month'
and date_info = '2024-03'
"""
df_hive = spark.sql(sql).repartition(40)
connection_info = StarRocksHelper.get_connection_info('selection')
df_hive.write.format("starrocks") \
.option("starrocks.fe.http.url", f"{connection_info['ip']}:{connection_info['http_port']}") \
.option("starrocks.fe.jdbc.url", f"jdbc:mysql://{connection_info['ip']}:{connection_info['jdbc_port']}") \
.option("starrocks.table.identifier", "test.ods_asin_detail_test") \
.option("starrocks.user", connection_info['user']) \
.option("starrocks.password", connection_info['pwd']) \
.option("starrocks.write.flush.interval.ms", "10000") \
.option("starrocks.write.properties.column_separator", "~!@#$%^&*~!@#$%^&*") \
.mode("append") \
.save()
print("导出完毕")
spark.stop()
Pyspark_job/ct/re_run_aba_2022.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.DolphinschedulerHelper import DolphinschedulerHelper
from utils.common_util import CommonUtil
if __name__ == '__main__':
date_list = ['2022-02', '2022-03', '2022-04', '2022-05', '2022-06',
'2022-07', '2022-08', '2022-09', '2022-10', '2022-11', '2022-12']
for date_info in date_list:
print(f"当前执行的分区为：{date_info}")
success_flag = DolphinschedulerHelper.start_and_watch_process_instance(
"big_data_selection",
process_df_name='ABA品牌标签调整重跑_api',
startParams={
"site_name": "us",
"date_type": "month",
"date_info": date_info,
"wx_user": "chenyuanjie"
}
)
if success_flag:
continue
else:
CommonUtil.send_wx_msg(["chenyuanjie"], f"ABA品牌标签调整重跑_api {date_info} 执行失败")
break
CommonUtil.send_wx_msg(["chenyuanjie"], "ABA品牌标签调整重跑_api 2022年 执行结束")
pass
Pyspark_job/ct/re_run_aba_2023.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.DolphinschedulerHelper import DolphinschedulerHelper
from utils.common_util import CommonUtil
if __name__ == '__main__':
date_list = ['2023-11', '2023-12']
for date_info in date_list:
print(f"当前执行的分区为：{date_info}")
success_flag = DolphinschedulerHelper.start_and_watch_process_instance(
"big_data_selection",
process_df_name='ABA品牌标签调整重跑_api',
startParams={
"site_name": "us",
"date_type": "month",
"date_info": date_info,
"wx_user": "chenyuanjie"
}
)
if success_flag:
continue
else:
CommonUtil.send_wx_msg(["chenyuanjie"], f"ABA品牌标签调整重跑_api {date_info} 执行失败")
break
CommonUtil.send_wx_msg(["chenyuanjie"], "ABA品牌标签调整重跑_api 2023年 执行结束")
pass
Pyspark_job/ct/re_run_aba_2024.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.DolphinschedulerHelper import DolphinschedulerHelper
from utils.common_util import CommonUtil
if __name__ == '__main__':
date_list = ['2024-01', '2024-02', '2024-03', '2024-04',
'2024-05', '2024-06', '2024-07', '2024-08']
for date_info in date_list:
print(f"当前执行的分区为：{date_info}")
success_flag = DolphinschedulerHelper.start_and_watch_process_instance(
"big_data_selection",
process_df_name='ABA品牌标签调整重跑_api',
startParams={
"site_name": "us",
"date_type": "month",
"date_info": date_info,
"wx_user": "chenyuanjie"
}
)
if success_flag:
continue
else:
CommonUtil.send_wx_msg(["chenyuanjie"], f"ABA品牌标签调整重跑_api {date_info} 执行失败")
break
CommonUtil.send_wx_msg(["chenyuanjie"], "ABA品牌标签调整重跑_api 2024年 执行结束")
pass
Pyspark_job/ct/spark_pg2pg.py deleted 100644 → 0
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.functions import row_number, lit
from pyspark.sql.window import Window
from pyspark.sql.types import StringType, ArrayType
from urllib.parse import quote
from datetime import datetime
if __name__ == '__main__':
date_info = CommonUtil.get_sys_arg(1, None)
year, month, day = date_info.split("-")
table = f"us_merchantwords_brand_analytics_2024_{month}_{day}"
spark = SparkUtil.get_spark_session(f"us_merchantwords_brand_analytics_2024:pg2pg,{date_info}")
df = spark.read.format("jdbc") \
.option("url", "jdbc:postgresql://113.100.143.162:5432/selection") \
.option("dbtable", table) \
.option("user", "yswg_postgres") \
.option("password", "yswg_postgres") \
.load()
df.write.format("jdbc") \
.option("url", "jdbc:postgresql://113.100.143.162:5443/selection") \
.option("dbtable", table) \
.option("user", "yswg_postgres") \
.option("password", "yswg_postgres") \
.mode("append") \
.save()
spark.stop()
\ No newline at end of file
Pyspark_job/ct/supplement_merchantwords_de.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.spark_util import SparkUtil
from pyspark.sql.functions import lit, col
from pyspark.sql.types import StringType, ArrayType
from urllib.parse import quote
if __name__ == '__main__':
export_tb = "de_merchantwords_search_term_month_syn_2024"
spark = SparkUtil.get_spark_session("MerchantwordsSupplement")
# 构建 URL 的函数
def build_urls(search_term):
url_template = f"https://www.amazon.de/s?k={{search_term}}&page={{page_number}}"
search_term_chinese = quote(search_term, 'utf-8')
search_term_chinese = search_term_chinese.replace("'", '%27').replace("/", '%2F')
urls = [
url_template.format(
search_term=search_term_chinese.replace(' ', '+').replace('&', '%26').replace('#', '%23').replace('(',
'%28').replace(
')', '%29'), page_number=1),
url_template.format(
search_term=search_term_chinese.replace(' ', '+').replace('&', '%26').replace('#', '%23').replace('(',
'%28').replace(
')', '%29'), page_number=2),
url_template.format(
search_term=search_term_chinese.replace(' ', '+').replace('&', '%26').replace('#', '%23').replace('(',
'%28').replace(
')', '%29'), page_number=3)
]
return urls
# 将Python函数转换为UDF
spark.udf.register("build_urls", build_urls, ArrayType(StringType()))
sql1 = """
select
keyword,
volume,
st_monthly_sales,
greatest(results_count, asin_total_num) as asin_total_num,
st_sp_counts,
st_zr_counts
from dwt_merchantwords_merge
where site_name = 'de'
and batch = '2024-07-01'
"""
df_dwt_merchantwords_merge = spark.sql(sql1)
# sql2 = """
# select
# keyword
# from dwt_merchantwords_st_detail
# where site_name = 'de'
# and batch = '2024-1'
# """
# df_dwt_merchantwords_st_detail = spark.sql(sql2)
# 产品总数大于80且没有月销
df1 = df_dwt_merchantwords_merge.filter('asin_total_num > 80 and st_monthly_sales <= 0').select('keyword')
print("产品总数大于80且没有月销:" + str(df1.count()))
# 搜索量较大且没有sp广告词
df2 = df_dwt_merchantwords_merge.filter('volume >= 1 and st_sp_counts <= 0').select('keyword')
print("搜索量较大且没有sp广告词:" + str(df2.count()))
# 自然词总数 <= 0的部分
df3 = df_dwt_merchantwords_merge.filter('st_zr_counts <= 0').select('keyword')
print("自然词总数 <= 0的部分:" + str(df3.count()))
# # 过滤掉keyword含有中文的数据
# df_hive = df_hive.filter(~df_hive["keyword"].rlike("[\u4e00-\u9fff]"))
df_save = df1.union(df2).union(df3).drop_duplicates(['keyword'])
df_save = df_save.selectExpr("keyword AS search_term")
df_save = df_save.selectExpr("search_term", "explode(build_urls(search_term)) AS url")
df_save = df_save.withColumn("date_info", lit('2024-06-26'))
# 导出数据到 PostgreSQL 数据库
df_save.write.format("jdbc") \
.option("url", "jdbc:postgresql://192.168.10.225:5433/selection_de") \
.option("dbtable", export_tb) \
.option("user", "yswg_postgres") \
.option("password", "yswg_postgres") \
.mode("append") \
.save()
spark.stop()
\ No newline at end of file
Pyspark_job/ct/supplement_merchantwords_us.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.spark_util import SparkUtil
from pyspark.sql.functions import lit, col
from pyspark.sql.types import StringType, ArrayType
from urllib.parse import quote
if __name__ == '__main__':
export_tb = "us_merchantwords_search_term_month_syn_2024"
spark = SparkUtil.get_spark_session("MerchantwordsSupplement")
# 构建 URL 的函数
def build_urls(search_term):
url_template = f"https://www.amazon.com/s?k={{search_term}}&page={{page_number}}"
search_term_chinese = quote(search_term, 'utf-8')
search_term_chinese = search_term_chinese.replace("'", '%27').replace("/", '%2F')
urls = [
url_template.format(
search_term=search_term_chinese.replace(' ', '+').replace('&', '%26').replace('#', '%23').replace('(',
'%28').replace(
')', '%29'), page_number=1),
url_template.format(
search_term=search_term_chinese.replace(' ', '+').replace('&', '%26').replace('#', '%23').replace('(',
'%28').replace(
')', '%29'), page_number=2),
url_template.format(
search_term=search_term_chinese.replace(' ', '+').replace('&', '%26').replace('#', '%23').replace('(',
'%28').replace(
')', '%29'), page_number=3)
]
return urls
# 将Python函数转换为UDF
spark.udf.register("build_urls", build_urls, ArrayType(StringType()))
sql1 = """
select
keyword,
volume,
st_zr_counts,
st_sp_counts
from dwt_merchantwords_merge
where site_name = 'us'
and batch = '2024-07-01'
"""
df_dwt_merchantwords_merge = spark.sql(sql1)
# 搜索量较大且没有sp广告词
df1 = df_dwt_merchantwords_merge.filter('volume >= 1 and st_sp_counts <= 0').select('keyword')
print("搜索量较大且没有sp广告词:" + str(df1.count()))
# 自然词总数 <= 0的部分
df2 = df_dwt_merchantwords_merge.filter('st_zr_counts <= 0').select('keyword')
print("自然词总数 <= 0的部分:" + str(df2.count()))
# # 过滤掉keyword含有中文的数据
# df_hive = df_hive.filter(~df_hive["keyword"].rlike("[\u4e00-\u9fff]"))
df_save = df1.union(df2).drop_duplicates(['keyword'])
df_save = df_save.selectExpr("keyword AS search_term")
df_save = df_save.selectExpr("search_term", "explode(build_urls(search_term)) AS url")
df_save = df_save.withColumn("date_info", lit('2024-06-26'))
# 导出数据到 PostgreSQL 数据库
df_save.write.format("jdbc") \
.option("url", "jdbc:postgresql://192.168.10.225:5433/selection") \
.option("dbtable", export_tb) \
.option("user", "yswg_postgres") \
.option("password", "yswg_postgres") \
.mode("append") \
.save()
spark.stop()
\ No newline at end of file
Pyspark_job/ct/test.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from utils.DorisHelper import DorisHelper
from utils.spark_util import SparkUtil
from pyspark.sql import functions as F
if __name__ == '__main__':
spark = SparkUtil.get_spark_session('aba_to_doris_test')
sql = f"""
select *
from dwt_aba_last365
where site_name = 'us'
and date_type = 'month'
and date_info = '2024-10';
"""
df_aba = spark.sql(sql).drop('site_name', 'date_type').cache()
df_aba = df_aba.withColumn(
'date_info', F.concat(F.regexp_replace('date_info', '-', ''), F.lit('01'))
)
df_aba.show(10, True)
columns = df_aba.columns
columns_str = ",".join(columns)
DorisHelper.spark_export_with_columns(df_aba, 'test', 'dwt_aba_last365', columns_str)
print('导出完成')
Pyspark_job/ct/test1.py deleted 100644 → 0
from openai import OpenAI
api_key = "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"
client = OpenAI(api_key=api_key)
completion = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{
"role": "user",
"content": "Write a haiku about recursion in programming."
}
]
)
print(completion.choices[0].message)
Pyspark_job/ct/test2.py deleted 100644 → 0
import requests
response = requests.post(
f"https://api.stability.ai/v2beta/stable-image/generate/ultra",
headers={
"authorization": f"sk-f2iOAkResIloOY3yE6xk2LlQbVrtQi3EczZDjA3n9ns7bmeR",
"accept": "image/*"
},
files={"none": ''},
data={
"prompt": "A little cat is in a bedroom with a bed, TV, and sofa",
"output_format": "webp",
},
)
if response.status_code == 200:
with open("./cat01.webp", 'wb') as file:
file.write(response.content)
else:
raise Exception(str(response.json()))
Pyspark_job/ct/test_translate.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.common_util import CommonUtil
from utils.hdfs_utils import HdfsUtils
from utils.spark_util import SparkUtil
from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, StringType, StructType, StructField, BooleanType, MapType
"""
merchantwords 搜索词分词词频
"""
def is_number(str):
"""
判断一个字符是否是数字
:param str:
:return:
"""
import re
return re.match(r"^-?\d+\.?\d+$", str) is not None
def word_tokenize(keyword: str):
import re
keyword = re.sub(r'(\d+\.?\d*|-|\"|,|，|？|\?|/|、|)', '', keyword).strip()
from nltk.tokenize import word_tokenize
result = word_tokenize(keyword, "english")
# 过滤标点如下
filter_arr = [
" ", "\t", "\r", "\n", "(", ")", ",", "，", "[", "]", "、", "-", ":", "&", "|", "+", "``", "'", "'", "\""
]
return list(filter(lambda x: not is_number(x) and x not in filter_arr, result))
def run():
spark = SparkUtil.get_spark_session("app_name")
udf_word_tokenize = F.udf(word_tokenize, ArrayType(StringType()))
keywords_all = spark.sql("select keyword from dwt_merchantwords_st_detail where site_name='us'").cache()
df_all = keywords_all.withColumn("word", F.explode(udf_word_tokenize(F.col("keyword"))))
df_all = df_all.groupby(F.col("word")) \
.agg(F.count("word").alias("frequency")) \
.orderBy(F.col("frequency").desc()) \
.select(
F.col("word"),
F.col("frequency"),
F.lit("us").alias("site_name")
)
hive_tb = 'tmp_word_frequency'
# # 去重
partition_dict = {
"site_name": "us"
}
hdfs_path = CommonUtil.build_hdfs_path(hive_tb, partition_dict)
HdfsUtils.delete_hdfs_file(hdfs_path)
partition_by = list(partition_dict.keys())
print(f"当前存储的表名为：{hive_tb},分区为{partition_by}", )
df_all.write.saveAsTable(name=hive_tb, format='hive', mode='append', partitionBy=partition_by)
def word_pluralize(keyword: str):
from textblob import Word
# 单数形式
singularize = Word(keyword).singularize().string
# 复数形式
pluralize = Word(singularize).pluralize().string
result = {
"text": keyword,
"singularize": singularize,
"pluralize": pluralize,
"pluralizeFlag": keyword == pluralize,
"not_regular": keyword not in [singularize, pluralize]
}
return result
def word_stem(keyword: str):
from nltk.stem.snowball import SnowballStemmer
stemmer = SnowballStemmer("english", ignore_stopwords=False)
return stemmer.stem(keyword)
def word_test():
spark = SparkUtil.get_spark_session("word_test")
udf_word_pluralize = F.udf(word_pluralize, StructType(
[
StructField('text', StringType(), True),
StructField('singularize', StringType(), True),
StructField('pluralize', StringType(), True),
StructField('pluralizeFlag', BooleanType(), True),
StructField('not_regular', BooleanType(), True),
]
))
udf_word_stem = F.udf(word_stem, StringType())
keywords_all = spark.sql("select word,frequency from tmp_word_frequency").cache()
keywords_all = keywords_all.withColumn("resultMap", udf_word_pluralize(F.col("word"))).select(
F.col("word"),
F.col("frequency"),
F.col("resultMap").getField("singularize").alias("singularize"),
F.col("resultMap").getField("pluralize").alias("pluralize"),
F.col("resultMap").getField("pluralizeFlag").alias("pluralizeFlag"),
F.col("resultMap").getField("not_regular").alias("not_regular"),
).where("(pluralizeFlag == true) or (not_regular == true)")
# 计算词根
keywords_all = keywords_all.withColumn("word_stem", udf_word_stem(F.col("word")))
keywords_all = keywords_all.withColumn("singularize_stem", udf_word_stem(F.col("singularize")))
keywords_all = keywords_all.withColumn("pluralize_stem", udf_word_stem(F.col("pluralize")))
hive_tb = 'tmp_word_not_regular_v2'
keywords_all.write.saveAsTable(name=hive_tb, format='hive', mode='append')
print("success")
def word_for_download():
spark = SparkUtil.get_spark_session("word_for_calc")
keywords_all = spark.sql("""
select word
from tmp_for_market
order by volume desc
""")
CommonUtil.df_export_csv(spark, keywords_all, csv_name='word_for_calc', limit=200 * 10000)
print("success")
pass
if __name__ == '__main__':
# word_for_calc()
word_for_download()
print("success")
Pyspark_job/ct/test_udf.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from pyspark.sql.types import StringType
from utils.templates import Templates
from google.cloud import translate_v2 as translate
class Test(Templates):
def __init__(self):
super().__init__()
self.spark = self.create_spark_object(app_name=f"test")
self.df_st = self.spark.sql(f"select 1+1;")
self.translate_client = translate.Client()
# 自定义udf
self.u_translate_text = self.spark.udf.register('translate_text', self.translate_text, StringType())
def translate_text(self, word: str, target_language='zh'):
result = self.translate_client.translate(word, target_language=target_language)
return result['translatedText']
def read_data(self):
sql1 = f"""
select
search_term
from dwt_aba_last365
where site_name = 'us'
and date_type = 'last365day'
and date_info = '2023-12';
"""
self.df_st = self.spark.sql(sql1).limit(20).cache()
def handle_data(self):
self.df_st = self.df_st.withColumn(
'translate_text',
self.u_translate_text(self.df_st['search_term'])
)
self.df_st.show(20, False)
if __name__ == '__main__':
handle_obj = Test()
handle_obj.run()
Pyspark_job/ct/test_vertify.py deleted 100644 → 0
import os
import sys
import json
sys.path.append(os.path.dirname(sys.path[0]))
from utils.db_util import DBUtil
from utils.common_util import CommonUtil
from utils.spark_util import SparkUtil
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
if __name__ == '__main__':
site_name = CommonUtil.get_sys_arg(1, None)
date_type = CommonUtil.get_sys_arg(2, None)
date_info = CommonUtil.get_sys_arg(3, None)
assert site_name is not None, "site_name 不能为空！"
assert date_type is not None, "date_type 不能为空！"
assert date_info is not None, "date_info 不能为空！"
hive_table = f"dwt_flow_asin"
partition_dict = {
"site_name": site_name,
"date_type": date_type,
"date_info": date_info
}
# 获取计算分区
msg_params = ""
# 解析partition_dict获取分区查询条件
partition_conditions = []
for key, value in partition_dict.items():
if value is not None:
msg_params += f"{value} "
partition_conditions.append(f"{key} = '{value}'")
base_msg = f"{hive_table} {msg_params} "
site_name = partition_dict.get("site_name")
date_type = partition_dict.get("date_type")
spark_session = SparkUtil.get_spark_sessionV3("check_fields_rule")
# 获取维护的字段验证配置表数据
config_table_query = f"""select * from hive_field_verify_config
where table_name ='{hive_table}'
and site_name = '{site_name}'
and use_flag = 1 """
conn_info = DBUtil.get_connection_info('postgresql', 'us')
check_field_df = SparkUtil.read_jdbc_query(
session=spark_session,
url=conn_info["url"],
pwd=conn_info["pwd"],
username=conn_info["username"],
query=config_table_query
)
# 获取验证消息
check_field_list = check_field_df.select('field_name', 'verify_desc', 'verify_type', 'config_json',
'msg_usr_list').collect()
if not check_field_list:
print("============================无验证匹配条件跳过验证===================================")
exit()
# 创建一个df用于储存验证情况
# 定义列的结构
schema = StructType([
StructField("验证描述", StringType(), True),
StructField("验证类型", StringType(), True),
StructField("校验字段", StringType(), True),
StructField("校验条件查询占比", StringType(), True),
StructField("验证占比临界值上限", StringType(), True),
StructField("验证占比临界值下限", StringType(), True),
StructField("是否验证通过", IntegerType(), True),
])
# 使用定义的结构创建空的 DataFrame
check_df = spark_session.createDataFrame([], schema)
# 进行验证sql组装
query = f"""
SELECT COUNT(1) AS total_count
FROM {hive_table}
"""
# 拼接where条件
if partition_conditions:
query_total = query + f" WHERE {' AND '.join(partition_conditions)}"
# 执行sql获取验证值与df
total_df = spark_session.sql(query_total).cache()
total_count = int(total_df.collect()[0]['total_count'])
for row in check_field_list:
vertify_flag = True
field_name = row['field_name']
verify_type = row['verify_type']
config_json = json.loads(row['config_json'])
msg_usr = row['msg_usr_list']
msg_usr_list = [user.strip() for user in msg_usr.split(",")] if msg_usr else []
sql_condition = config_json['sql_condition']
partition_conf_list = config_json['partition_conf']
for conf in partition_conf_list:
conf_site_name = conf["site_name"]
conf_date_type = conf["date_type"]
if site_name == conf_site_name and date_type == conf_date_type:
vertify_flag = True
break
else:
vertify_flag = False
# 没有合适的匹配维度
if not vertify_flag:
break
# 拼接外部查询条件
if sql_condition:
query_field_check = query_total + f" AND {sql_condition} "
check_count_df = spark_session.sql(query_field_check).cache()
check_count = int(check_count_df.collect()[0]['total_count'])
calcult_rate = round((check_count / total_count), 3)
waring_max = conf['max_rate']
waring_min = conf['min_rate']
verify_flag = 1 if (calcult_rate <= waring_max) and (calcult_rate >= waring_min) else 0
ratio_df = spark_session.createDataFrame([(row['verify_desc'],verify_type,field_name,calcult_rate,waring_max,waring_min,verify_flag)],schema).repartition(1)
check_df = check_df.unionByName(ratio_df, False)
if check_df.count() < 1 :
print("无验证项验证")
exit()
check_df.show(50, truncate=False)
# 对校验结果进行判断是否有校验不通过的数据
schema_flag = bool(check_df.select(F.min("是否验证通过").alias("result")).first().asDict()['result'])
if not schema_flag:
msg = f"数据表：{hive_table} {msg_params}，计算数据存在验证不通过，请检查数据是否异常！！具体信息请查看日志！！"
CommonUtil.send_wx_msg(['chenjianyun'], f"\u26A0 {hive_table} {msg_params}流程数据导出前验证异常", msg)
spark_session.stop()
pass
\ No newline at end of file
Pyspark_job/ct/test_word_tokenize.py deleted 100644 → 0
def word_tokenize(title: str):
"""
分词器
"""
from nltk.tokenize import word_tokenize
result = word_tokenize(title, "english")
return result
if __name__ == '__main__':
aba = "nation's bravest tales of courage and heroism"
print(word_tokenize(aba))
Pyspark_job/ct/update_de_brand_analytics_month_2024_05.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.spark_util import SparkUtil
if __name__ == '__main__':
export_tb = "de_brand_analytics_month"
spark = SparkUtil.get_spark_session("update_de_brand_analytics_month_2024_05")
sql1 = """
select
search_term
from ods_st_quantity_being_sold
where site_name = 'de'
and date_type = 'month'
and date_info = '2024-05'
and quantity_being_sold in (16, 48)
"""
df_aba = spark.sql(sql1)
sql2 = """
select
search_term,
quantity_being_sold
from dwt_merchantwords_merge
where site_name = 'de'
"""
df_me = spark.sql(sql2)
df_save = df_aba.join(
df_me, on='search_term', how='inner'
)
# 导出数据到 PostgreSQL 数据库
df_save.write.format("jdbc") \
.option("url", "jdbc:postgresql://192.168.10.223:5433/selection_de") \
.option("dbtable", export_tb) \
.option("user", "yswg_postgres") \
.option("password", "yswg_postgres") \
.mode("append") \
.save()
spark.stop()
\ No newline at end of file
Pyspark_job/ct/update_merchantwords_measure.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from pyspark.sql.functions import row_number
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from utils.common_util import CommonUtil
from utils.spark_util import SparkUtil
if __name__ == '__main__':
spark = SparkUtil.get_spark_session("UpdateMerchantwords")
hive_tb = 'dwd_merchantwords_measure'
partition_dict = {
"site_name": 'us',
"batch": '2023-01'
}
sql1 = f"""
select
keyword,
lang,
st_ao_val,
st_zr_flow_proportion,
min_bid,
max_bid,
suggested_bid,
volume,
avg_3m,
avg_12m,
asin_total_num,
asin_num,
self_asin_num,
self_asin_proportion,
st_sp_counts,
st_zr_counts,
st_monthly_sales,
listing_sales_avg,
reviews_avg,
rating_avg,
price_avg,
depth
from dwd_merchantwords_measure
where site_name = 'us'
and batch = '2023-01';
"""
df_dwd = spark.sql(sqlQuery=sql1).cache()
df_dwd.repartition(80)
sql2 = f"""
select
keyword,
results_count,
sponsored_ads_count,
page_1_reviews,
appearance,
last_seen,
update_time
from dwt_merchantwords_st_detail
where site_name = 'us'
and batch = '2023-1';
"""
df_merchantwords_detail = spark.sql(sqlQuery=sql2)
df_merchantwords_detail = df_merchantwords_detail \
.withColumn("row_num", row_number().over(Window.orderBy("keyword"))) \
.filter("row_num BETWEEN 1 AND 12000000") \
.repartition(80) \
.drop("row_num") \
.cache()
df = df_dwd.join(df_merchantwords_detail, 'keyword', 'left')
df = df.withColumn(
'site_name',
F.lit('us')
).withColumn(
'batch',
F.lit('2023-01')
)
CommonUtil.save_or_update_table(spark_session=spark,
hive_tb_name=hive_tb,
partition_dict=partition_dict,
df_save=df,
drop_exist_tmp_flag=True)
Pyspark_job/ct/us_aba_last365_word_frequency.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.spark_util import SparkUtil
from pyspark.sql.functions import count, col
class WordFrequency(object):
def __init__(self):
self.spark = SparkUtil.get_spark_session("us_aba_last365_word_frequency")
def run(self):
sql1 = f"""
select search_term, date_info
from dwt_aba_st_analytics
where site_name = 'us'
and date_type = 'month'
and date_info in
('2024-10', '2024-09', '2024-08', '2024-07', '2024-06', '2024-05',
'2024-04', '2024-03', '2024-02', '2024-01', '2023-12', '2023-11')
and rank <= 1000000
and st_brand_label = 1;
"""
df_st = self.spark.sql(sql1).cache()
print("df_st数量是：")
print(df_st.count())
sql2 = f"""
select search_term, first_match_brand as brand, date_info
from dws_st_brand_info
where site_name = 'us'
and date_type = 'month'
and date_info in
('2024-10', '2024-09', '2024-08', '2024-07', '2024-06', '2024-05',
'2024-04', '2024-03', '2024-02', '2024-01', '2023-12', '2023-11')
and st_brand_label = 1;
"""
df_brand = self.spark.sql(sql2).cache()
print("df_brand数量是：")
print(df_brand.count())
df_save = df_st.join(
df_brand, on=['date_info', 'search_term'], how='left'
).drop('date_info')
print("df_save数量是：")
print(df_save.count())
df_save = df_save.groupby(['brand']).agg(
count('brand').alias('frequency')
).orderBy('frequency', ascending=False)
df_save.show(20, False)
df_save = df_save.withColumn("frequency", col("frequency").cast("int"))
total_sum = df_save.select("frequency").groupBy().sum().collect()[0][0]
if total_sum == df_st.count():
print('验证成功')
else:
print('验证失败')
output_path = "hdfs:///user/chenyuanjie/test1/"
df_save.write.mode("overwrite").format("csv").option("delimiter", "^").option("lineSep", "\n").option("header", "false").option("compression", "none").save(output_path)
if __name__ == '__main__':
obj = WordFrequency()
obj.run()
Pyspark_job/ct/us_st_keepa_syn_2024.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.ssh_util import SSHUtil
from utils.common_util import CommonUtil
from utils.hdfs_utils import HdfsUtils
if __name__ == '__main__':
hive_tb = "tmp_us_st_keepa_syn_2024"
hdfs_path = "/home/big_data_selection/tmp/tmp_us_st_keepa_syn_2024"
print(f"hdfs_path is {hdfs_path}")
query = f"""
select
asin
from us_st_keepa_syn_2024
where 1 = 1
and \$CONDITIONS
"""
db_type = "postgresql"
empty_flag, check_flag = CommonUtil.check_schema_before_import(db_type=db_type,
site_name='us',
query=query,
hive_tb_name=hive_tb,
msg_usr=['chenyuanjie']
)
assert check_flag, f"导入hive表{hive_tb}表结构检查失败！请检查query是否异常！！"
if not empty_flag:
sh = CommonUtil.build_import_sh(site_name='us',
db_type=db_type,
query=query,
hdfs_path=hdfs_path)
# 导入前先删除
HdfsUtils.delete_hdfs_file(hdfs_path)
client = SSHUtil.get_ssh_client()
SSHUtil.exec_command_async(client, sh, ignore_err=False)
CommonUtil.after_import(hdfs_path=hdfs_path, hive_tb=hive_tb)
client.close()
pass
Pyspark_job/ct/us_st_keepa_syn_2024_export.py deleted 100644 → 0
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.spark_util import SparkUtil
from pyspark.sql.functions import col, lit
from utils.StarRocksHelper import StarRocksHelper
if __name__ == '__main__':
spark = SparkUtil.get_spark_session("us_st_keepa_syn_2024_export")
# 从SR数据库中读取已有数据
sql = """
select distinct asin from selection.us_asin_latest_detail where date_info = '2024-06'
"""
df_sr = StarRocksHelper.spark_import_with_sql(spark, sql).repartition(80, 'asin').cache()
print("starrocks读取：")
df_sr.show(10)
sql = """
select asin from tmp_us_st_keepa_syn_2024;
"""
df_pg = spark.sql(sql).drop_duplicates(['asin']).repartition(80, 'asin').cache()
print("pg读取：")
df_pg.show(10)
df = df_sr.subtract(df_pg)
df_sr.unpersist()
df_pg.unpersist()
df = df.withColumn(
'state',
lit(7)
).withColumn(
'asin_trun_4',
col('asin').substr(1, 4)
)
df.show(10)
print(df.count())
df.write.format("jdbc") \
.option("url", "jdbc:postgresql://192.168.10.224:5433/selection") \
.option("dbtable", "us_st_keepa_syn_2024") \
.option("user", "yswg_postgres") \
.option("password", "yswg_postgres") \
.mode("append") \
.save()
spark.stop()
Pyspark_job/ct/us_st_keepa_syn_2024_update.py deleted 100644 → 0
import os
import sys
from sqlalchemy.dialects.postgresql import pypostgresql
sys.path.append(os.path.dirname(sys.path[0]))
from utils.spark_util import SparkUtil
from utils.db_util import DBUtil
from utils.StarRocksHelper import StarRocksHelper
if __name__ == '__main__':
spark = SparkUtil.get_spark_session("us_st_keepa_syn_2024_export")
# 从SR数据库中读取已有数据
sql = """
select distinct asin from selection.us_asin_latest_detail where date_info = '2024-06' and (asin_launch_time>'2024-07-19' or asin_launch_time<'1990-01-01')
"""
df_sr = StarRocksHelper.spark_import_with_sql(spark, sql).repartition(80, 'asin').cache()
print("starrocks读取：")
df_sr.show(10)
sql = """
select asin from tmp_us_st_keepa_syn_2024;
"""
df_pg = spark.sql(sql).drop_duplicates(['asin']).repartition(80, 'asin').cache()
print("pg读取：")
df_pg.show(10)
df = df_sr.subtract(df_pg)
print(df.count())
df_sr.unpersist()
df_pg.unpersist()
update_asin = df.select("asin").rdd.map(lambda row: row[0]).collect()
print(update_asin)
pg_engine = DBUtil.get_db_engine('postgresql', 'us')
with pg_engine.begin() as conn:
update_query = f"""
UPDATE us_st_keepa_syn_2024 SET state = 5 WHERE asin IN {tuple(update_asin)}
"""
conn.execute(update_query)
spark.stop()
Pyspark_job/ct/verify_rank.py deleted 100644 → 0
import os
import sys
from pyspark.sql.types import ArrayType, FloatType, StructType, StructField, StringType
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
class VerifyRank(object):
def __init__(self):
self.spark = SparkUtil.get_spark_session("{self.__class__.__name__}")
def run(self):
sql = f"""
select
search_term,
rank,
date_info
from ods_brand_analytics
where site_name = 'us'
and date_type = 'week'
and date_info >= '2024-01'
and rank < 100000
"""
df_all = self.spark.sql(sql).repartition(40, 'search_term').cache()
def leave_one_out_means(structs):
ranks = [x['rank'] for x in structs]
date_infos = [x['date_info'] for x in structs]
total_sum = sum(ranks)
n = len(ranks)
if n > 1:
means = [round((total_sum - rank) / (n - 1), 2) for rank in ranks]
else:
means = [ranks[0]]
result = [{"means": mean, "date_info": date_info} for mean, date_info in zip(means, date_infos)]
return result
leave_one_out_means_udf = F.udf(leave_one_out_means, ArrayType(StructType([
StructField("means", FloatType(), True),
StructField("date_info", StringType(), True)
])))
df_agg = df_all.groupBy("search_term").agg(
F.collect_list(F.struct("rank", "date_info")).alias("collect_row")
# F.collect_list("rank").alias("values")
)
df_agg = df_agg.withColumn(
"collect_row", leave_one_out_means_udf(F.col("collect_row"))
)
def calc_quantiles(structs):
values = [x['means'] for x in structs]
values = sorted(values) # 将组内的数值进行排序
n = len(values)
# 计算 Q1 和 Q3 的位置（基于 25% 和 75% 的位置）
q1_index = int(n * 0.25)
q3_index = int(n * 0.75)
if n > 1:
q1 = values[q1_index]
q3 = values[q3_index]
else:
q1 = values[0]
q3 = values[0]
return [float(q1), float(q3)]
quantile_udf = F.udf(calc_quantiles, ArrayType(FloatType()))
df_agg = df_agg.withColumn(
"quantiles", quantile_udf(F.col("collect_row"))
).withColumn(
"q1", F.col("quantiles")[0]
).withColumn(
"q3", F.col("quantiles")[1]
).withColumn(
"iqr", F.expr("q3 - q1")
).withColumn(
"lower_bound", F.expr("q1 - 100 * iqr")
).withColumn(
"upper_bound", F.expr("q3 + 100 * iqr")
).select(
'search_term', 'collect_row', 'lower_bound', 'upper_bound'
).repartition(40, 'search_term')
df_save = df_agg.withColumn(
"filtered_collect_row",
F.filter(
"collect_row",
lambda x: (x["means"] < F.col("lower_bound")) | (x["means"] > F.col("upper_bound"))
)
).filter(
F.size(F.col("filtered_collect_row")) > 0
).withColumn(
"has_2024_08",
F.exists(
"filtered_collect_row",
lambda x: x["date_info"].like("2024-08%")
)
).filter(
~F.col("has_2024_08") # 过滤掉包含 '2024-08' 的行
).select(
'search_term', 'filtered_collect_row', 'lower_bound', 'upper_bound'
)
df_save.show(20, False)
if __name__ == '__main__':
site_name = CommonUtil.get_sys_arg(1, None)
date_type = CommonUtil.get_sys_arg(2, None)
date_info = CommonUtil.get_sys_arg(3, None)
obj = VerifyRank()
obj.run()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment