Skip to content

A
Amazon-Selection-Data
Commit 75c5c3ae by chenyuanjie
Options

etsy店铺数据同步

parent c9eef618
Showing with 428 additions and 0 deletions
Pyspark_job/sqoop_export/ods_ehunt_store_detail_pg.py 0 → 100644
"""
@Author : CT
@Description : eHunt店铺详情 hive ods_ehunt_store_detail -> pg集群 ehunt_store_detail_{yyyy}(按date_info分区的分布式表)
"""
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from sqlalchemy import text
from utils.db_util import DBUtil
from utils.ssh_util import SSHUtil
from utils.common_util import CommonUtil, DateTypes
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)
# 获取最后一个参数
test_flag = CommonUtil.get_sys_arg(len(sys.argv) - 1, None)
print(f"执行参数为{sys.argv}")
assert date_type == DateTypes.month.name, "date_type类型不对，应为month"
CommonUtil.judge_is_work_hours(site_name=site_name, date_type=date_type, date_info=date_info,
principal='chenyuanjie',
priority=1, export_tools_type=1, belonging_to_process='店铺流程')
if test_flag == 'test':
db_type = 'postgresql_test'
print("导出到测试库中")
else:
db_type = "postgresql_cluster"
print("导出到PG集群库中")
year_str = CommonUtil.safeIndex(date_info.split("-"), 0, None)
suffix = str(date_info).replace("-", "_")
base_tb = "ehunt_store_detail"
export_master_tb = f"{base_tb}_{year_str}"
export_master_tb_before = f"{base_tb}_{int(year_str) - 1}"
export_tb = f"{base_tb}_{suffix}"
export_tb_tmp = f"{export_tb}_tmp"
next_month = CommonUtil.get_next_val(date_type, date_info)
engine = DBUtil.get_db_engine(db_type, site_name)
# 当年母表不存在时用上一年母表 like 出来
with engine.connect() as conn:
master_exists = conn.execute(text(f"select to_regclass('{export_master_tb}') is not null")).scalar()
before_exists = conn.execute(text(f"select to_regclass('{export_master_tb_before}') is not null")).scalar()
if not master_exists:
assert before_exists, f"母表{export_master_tb}与{export_master_tb_before}均不存在，请先手动建表！"
DBUtil.engine_exec_sql(engine, f"""
create table if not exists {export_master_tb}
(like {export_master_tb_before} including all)
partition by range (date_info);
""")
# like 不带分布式属性，母表非分布式时按单副本转为分布式
if db_type == 'postgresql_cluster':
with engine.connect() as conn:
is_dist = conn.execute(text(f"select count(1) from pg_dist_partition where logicalrelid::text = '{export_master_tb}'")).scalar() > 0
if not is_dist:
with engine.begin() as conn:
conn.execute(text("set local citus.shard_replication_factor = 1"))
conn.execute(text(f"select create_distributed_table('{export_master_tb}', 'store_name')"))
# 当月子表直接建在母表下；中间表字段与hive一致，sqoop先导入中间表
sql = f"""
create table if not exists {export_tb} partition of {export_master_tb}
for values from ('{date_info}') to ('{next_month}');
drop table if exists {export_tb_tmp};
create table {export_tb_tmp}
(
store_name varchar,
category varchar,
country varchar,
sales int4,
sales_weekly int4,
revenue_total float8,
revenue_7days float8,
reviews int4,
reviews_weekly int4,
favorites int4,
favorites_weekly int4,
products int4,
start_at varchar,
is_star varchar,
is_raving varchar,
rating float8,
store_url varchar,
social_info varchar,
pay_type varchar,
date_info varchar,
created_at varchar
);
"""
DBUtil.engine_exec_sql(engine, sql)
sh = CommonUtil.build_export_sh(
site_name=site_name,
db_type=db_type,
hive_tb="ods_ehunt_store_detail",
export_tb=export_tb_tmp,
col=[
"store_name",
"category",
"country",
"sales",
"sales_weekly",
"revenue_total",
"revenue_7days",
"reviews",
"reviews_weekly",
"favorites",
"favorites_weekly",
"products",
"start_at",
"is_star",
"is_raving",
"rating",
"store_url",
"social_info",
"pay_type",
"date_info",
"created_at"
],
partition_dict={
"site_name": site_name,
"date_type": date_type,
"date_info": date_info
},
num_mappers=3
)
client = SSHUtil.get_ssh_client()
SSHUtil.exec_command_async(client, sh, ignore_err=False)
client.close()
# 同一事务内清空子表并从中间表写入，保证幂等；id按子表从1开始生成，分布列store_name不能为空
sql = f"""
truncate table {export_tb};
insert into {export_tb}
(id, store_name, category, country, sales, sales_weekly, revenue_total, revenue_7days, reviews, reviews_weekly,
favorites, favorites_weekly, products, start_at, is_star, is_raving, rating, store_url, social_info, pay_type,
date_info, created_at)
select row_number() over (order by store_name) as id, store_name, category, country, sales, sales_weekly, revenue_total::numeric, revenue_7days::numeric,
reviews, reviews_weekly, favorites, favorites_weekly, products, start_at, is_star, is_raving, rating::numeric,
store_url, social_info, pay_type, date_info, created_at::timestamp
from {export_tb_tmp}
where store_name is not null;
drop table if exists {export_tb_tmp};
"""
print(sql)
with engine.begin() as conn:
conn.execute(text(sql))
engine.dispose()
print("success")
Pyspark_job/sqoop_export/ods_ehunt_store_product_pg.py 0 → 100644
"""
@Author : CT
@Description : eHunt店铺商品 hive ods_ehunt_store_product -> pg集群 ehunt_store_product_{yyyy}(按date_info分区的分布式表)
"""
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from sqlalchemy import text
from utils.db_util import DBUtil
from utils.ssh_util import SSHUtil
from utils.common_util import CommonUtil, DateTypes
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)
# 获取最后一个参数
test_flag = CommonUtil.get_sys_arg(len(sys.argv) - 1, None)
print(f"执行参数为{sys.argv}")
assert date_type == DateTypes.month.name, "date_type类型不对，应为month"
CommonUtil.judge_is_work_hours(site_name=site_name, date_type=date_type, date_info=date_info,
principal='chenyuanjie',
priority=2, export_tools_type=1, belonging_to_process='店铺流程')
if test_flag == 'test':
db_type = 'postgresql_test'
print("导出到测试库中")
else:
db_type = "postgresql_cluster"
print("导出到PG集群库中")
year_str = CommonUtil.safeIndex(date_info.split("-"), 0, None)
suffix = str(date_info).replace("-", "_")
base_tb = "ehunt_store_product"
export_master_tb = f"{base_tb}_{year_str}"
export_master_tb_before = f"{base_tb}_{int(year_str) - 1}"
export_tb = f"{base_tb}_{suffix}"
export_tb_tmp = f"{export_tb}_tmp"
next_month = CommonUtil.get_next_val(date_type, date_info)
engine = DBUtil.get_db_engine(db_type, site_name)
# 当年母表不存在时用上一年母表 like 出来
with engine.connect() as conn:
master_exists = conn.execute(text(f"select to_regclass('{export_master_tb}') is not null")).scalar()
before_exists = conn.execute(text(f"select to_regclass('{export_master_tb_before}') is not null")).scalar()
if not master_exists:
assert before_exists, f"母表{export_master_tb}与{export_master_tb_before}均不存在，请先手动建表！"
DBUtil.engine_exec_sql(engine, f"""
create table if not exists {export_master_tb}
(like {export_master_tb_before} including all)
partition by range (date_info);
""")
# like 不带分布式属性，母表非分布式时按单副本转为分布式；默认按分布列类型与店铺详情表同分片
if db_type == 'postgresql_cluster':
with engine.connect() as conn:
is_dist = conn.execute(text(f"select count(1) from pg_dist_partition where logicalrelid::text = '{export_master_tb}'")).scalar() > 0
if not is_dist:
with engine.begin() as conn:
conn.execute(text("set local citus.shard_replication_factor = 1"))
conn.execute(text(f"select create_distributed_table('{export_master_tb}', 'store_name')"))
# 当月子表直接建在母表下；中间表字段与hive一致，sqoop先导入中间表
sql = f"""
create table if not exists {export_tb} partition of {export_master_tb}
for values from ('{date_info}') to ('{next_month}');
drop table if exists {export_tb_tmp};
create table {export_tb_tmp}
(
store_name varchar,
product_id varchar,
title varchar,
price float8,
sales_month int4,
product_url varchar,
logo_url varchar,
date_info varchar
);
"""
DBUtil.engine_exec_sql(engine, sql)
sh = CommonUtil.build_export_sh(
site_name=site_name,
db_type=db_type,
hive_tb="ods_ehunt_store_product",
export_tb=export_tb_tmp,
col=[
"store_name",
"product_id",
"title",
"price",
"sales_month",
"product_url",
"logo_url",
"date_info"
],
partition_dict={
"site_name": site_name,
"date_type": date_type,
"date_info": date_info
},
num_mappers=5
)
client = SSHUtil.get_ssh_client()
SSHUtil.exec_command_async(client, sh, ignore_err=False)
client.close()
# 同一事务内清空子表并从中间表写入，保证幂等；id按子表从1开始生成，分布列store_name不能为空
sql = f"""
truncate table {export_tb};
insert into {export_tb}
(id, store_name, product_id, title, price, sales_month, product_url, logo_url, date_info)
select row_number() over (order by store_name, product_id) as id, store_name, product_id::int8, title, price::numeric, sales_month, product_url, logo_url,
date_info
from {export_tb_tmp}
where store_name is not null;
drop table if exists {export_tb_tmp};
"""
print(sql)
with engine.begin() as conn:
conn.execute(text(sql))
engine.dispose()
print("success")
Pyspark_job/sqoop_import/ods_ehunt_store_detail.py 0 → 100644
"""
@Author : CT
@Description : eHunt店铺详情 pg14爬虫库 -> hive ods_ehunt_store_detail
"""
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.common_util import CommonUtil
from utils.common_util import DateTypes
from utils.secure_db_client import get_remote_engine
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 不能为空！"
assert date_type == DateTypes.month.name, "date_type类型不对，应为month"
suffix = str(date_info).replace("-", "_")
import_table = f"ehunt_store_detail_{suffix}"
db_type = 'postgresql_14'
print("当前链接的数据库为：", db_type)
sql_query = f"""
select
id,
store_name,
category,
country,
sales,
sales_weekly,
revenue_total,
revenue_7days,
reviews,
reviews_weekly,
favorites,
favorites_weekly,
products,
start_at,
is_star,
is_raving,
rating,
store_url,
social_info,
pay_type,
created_at
from {import_table}
where 1=1
and \$CONDITIONS
"""
hive_table = "ods_ehunt_store_detail"
partition_dict = {
"site_name": site_name,
"date_type": date_type,
"date_info": date_info
}
hdfs_path = CommonUtil.build_hdfs_path(hive_table, partition_dict=partition_dict)
engine = get_remote_engine(
site_name=site_name,
db_type=db_type
)
engine.sqoop_raw_import(
query=sql_query,
hive_table=hive_table,
hdfs_path=hdfs_path,
partitions=partition_dict
)
pass
Pyspark_job/sqoop_import/ods_ehunt_store_product.py 0 → 100644
"""
@Author : CT
@Description : eHunt店铺商品 pg14爬虫库 -> hive ods_ehunt_store_product
"""
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.common_util import CommonUtil
from utils.common_util import DateTypes
from utils.secure_db_client import get_remote_engine
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 不能为空！"
assert date_type == DateTypes.month.name, "date_type类型不对，应为month"
suffix = str(date_info).replace("-", "_")
import_table = f"ehunt_store_product_{suffix}"
db_type = 'postgresql_14'
print("当前链接的数据库为：", db_type)
sql_query = f"""
select
id,
store_name,
product_id,
regexp_replace(title, E'[\\r\\n\\t]+', ' ', 'g') as title,
price,
sales_month,
product_url,
logo_url
from {import_table}
where 1=1
and \$CONDITIONS
"""
hive_table = "ods_ehunt_store_product"
partition_dict = {
"site_name": site_name,
"date_type": date_type,
"date_info": date_info
}
hdfs_path = CommonUtil.build_hdfs_path(hive_table, partition_dict=partition_dict)
engine = get_remote_engine(
site_name=site_name,
db_type=db_type
)
engine.sqoop_raw_import(
query=sql_query,
hive_table=hive_table,
hdfs_path=hdfs_path,
partitions=partition_dict,
m=3,
split_by='id'
)
pass
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