Skip to content

A
Amazon-Selection-Data
Commit 350df446 by chenyuanjie
Options

kafka实时消费任务兼容爬虫流程

parent 4d60093f
Showing with 18 additions and 93 deletions
Pyspark_job/utils/templates.py
...@@ -71,8 +71,6 @@ class Templates(object): ...@@ -71,8 +71,6 @@ class Templates(object):
# my_kafka # my_kafka
self.consumer = object() self.consumer = object()
# 在此处定义 Kafka 认证和安全参数 # 在此处定义 Kafka 认证和安全参数
# self.kafka_servers = "192.168.10.221:9092,192.168.10.220:9092,192.168.10.210:9092"
# self.kafka_servers_producer = "61.145.136.61:19092,61.145.136.61:29092,61.145.136.61:39092"
self.kafka_servers = "192.168.10.218:9092,192.168.10.219:9092,192.168.10.220:9092" self.kafka_servers = "192.168.10.218:9092,192.168.10.219:9092,192.168.10.220:9092"
self.kafka_servers_producer = "'61.145.136.61:19092,61.145.136.61:29092,61.145.136.61:49092,61.145.136.61:59092'" self.kafka_servers_producer = "'61.145.136.61:19092,61.145.136.61:29092,61.145.136.61:49092,61.145.136.61:59092'"
self.kafka_security_protocol = "SASL_PLAINTEXT" self.kafka_security_protocol = "SASL_PLAINTEXT"
...@@ -93,8 +91,6 @@ class Templates(object): ...@@ -93,8 +91,6 @@ class Templates(object):
# 测试标识 # 测试标识
self.test_flag = 'normal' self.test_flag = 'normal'
self.beginning_offsets_dict = {} # history消费时, 初始的偏移量 self.beginning_offsets_dict = {} # history消费时, 初始的偏移量
# 记录最后一次收到非空批次的时间，用于无新数据超时检测
# self.last_data_time = time.time()
# 后台状态监控线程标记（防止重复启动） # 后台状态监控线程标记（防止重复启动）
self._state_monitor_started = False self._state_monitor_started = False
# kafka_stream_stop 防重入标记 # kafka_stream_stop 防重入标记
...@@ -170,7 +166,6 @@ class Templates(object): ...@@ -170,7 +166,6 @@ class Templates(object):
) )
return consumer return consumer
# @staticmethod
def get_kafka_partitions_data(self, consumer=None, topic_name="us_asin_detail"): def get_kafka_partitions_data(self, consumer=None, topic_name="us_asin_detail"):
partitions = consumer.partitions_for_topic(topic_name) partitions = consumer.partitions_for_topic(topic_name)
partition_data_count = {} partition_data_count = {}
...@@ -190,7 +185,6 @@ class Templates(object): ...@@ -190,7 +185,6 @@ class Templates(object):
"end_offsets": end_offsets, "end_offsets": end_offsets,
"data_count": data_count, "data_count": data_count,
} }
# partition_data_count[pid] = data_count
partition_data_count[pid] = offset_dict partition_data_count[pid] = offset_dict
print("partition_data_count:", partition_data_count) print("partition_data_count:", partition_data_count)
return partition_data_count return partition_data_count
...@@ -224,8 +218,6 @@ class Templates(object): ...@@ -224,8 +218,6 @@ class Templates(object):
self.year, self.month, self.day = self.date_info.split("-") self.year, self.month, self.day = self.date_info.split("-")
self.week = list(df_loc.year_week)[0].split("-")[-1] self.week = list(df_loc.year_week)[0].split("-")[-1]
if self.date_type in ['week', 'week_old', 'month', 'month_old']: if self.date_type in ['week', 'week_old', 'month', 'month_old']:
# df_loc = df.loc[df[f'year_{self.date_type}'] == f"{self.date_info}"]
# self.date_info_tuple = tuple(df_loc.date)
if self.date_type in ['week', 'week_old']: if self.date_type in ['week', 'week_old']:
df_loc = df.loc[df[f'year_week'] == f"{self.date_info}"] df_loc = df.loc[df[f'year_week'] == f"{self.date_info}"]
self.date_info_tuple = tuple(df_loc.date) self.date_info_tuple = tuple(df_loc.date)
...@@ -257,13 +249,11 @@ class Templates(object): ...@@ -257,13 +249,11 @@ class Templates(object):
def get_year_week_tuple(self): def get_year_week_tuple(self):
self.df_week = self.spark.sql(f"select * from dim_date_20_to_30 where week_day=1;") self.df_week = self.spark.sql(f"select * from dim_date_20_to_30 where week_day=1;")
# self.df_week = self.spark.sql(f"select * from dim_week_20_to_30;")
df = self.df_week.toPandas() df = self.df_week.toPandas()
df.year_month = df.year_month.apply(lambda x: x.replace("_", "-")) df.year_month = df.year_month.apply(lambda x: x.replace("_", "-"))
df.year_quarter = df.year_quarter.apply(lambda x: x.replace("_quarter_", "-")) df.year_quarter = df.year_quarter.apply(lambda x: x.replace("_quarter_", "-"))
if self.date_type in ['week']: if self.date_type in ['week']:
self.year_week = self.date_info self.year_week = self.date_info
# self.year, self.week = int(self.year_week.split("-")[0]), int(self.year_week.split("-")[1])
self.year, self.week = self.year_week.split("-")[0], self.year_week.split("-")[1] self.year, self.week = self.year_week.split("-")[0], self.year_week.split("-")[1]
self.year_week_tuple = f"('{self.year_week}')" self.year_week_tuple = f"('{self.year_week}')"
if self.date_type in ['4_week']: if self.date_type in ['4_week']:
...@@ -390,42 +380,6 @@ class Templates(object): ...@@ -390,42 +380,6 @@ class Templates(object):
t.start() t.start()
print(f"[状态监控] 后台监控线程已启动，每 {interval}s 检查一次爬虫状态") print(f"[状态监控] 后台监控线程已启动，每 {interval}s 检查一次爬虫状态")
# def kafka_consumption_is_finished(self):
# while True:
# try:
# # if self.site_name == 'us':
# # # sql = f"SELECT * from workflow_progress WHERE site_name='{self.site_name}' and page='{self.spider_type}' ORDER BY created_at desc LIMIT 1;"
# # sql = f"""
# # SELECT * from workflow_progress WHERE site_name='{self.site_name}' and page='{self.spider_type}'
# # and date_info in
# # -- (SELECT MAX(year_week) as date_info from date_20_to_30 WHERE `year_month` = '2024-02' and week_day =1
# # (SELECT year_week as date_info from date_20_to_30 WHERE `year_month` = '{self.date_info}' and week_day =1
# # )
# # ORDER BY created_at desc LIMIT 1;
# #
# # """
# # else:
# # sql = f"SELECT * from selection.workflow_progress WHERE site_name='{self.site_name}' and date_info='{self.date_info}' and page='{self.spider_type}' ORDER BY created_at desc LIMIT 1;"
# sql = f"SELECT * from selection.workflow_progress WHERE site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}' and page='{self.spider_type}' and spider_state=3;"
# print(f"判断爬虫'{self.spider_type}'是否结束, sql: {sql}")
# df = pd.read_sql(sql, con=self.engine_mysql)
# if df.shape[0]:
# status_val = list(df.status_val)[0]
# if int(status_val) == 3:
# print(f"spider_type:{self.spider_type}已经爬取完毕, 退出kafka消费和停止程序")
# if self.consumer_type == "latest":
# if HdfsUtils.delete_hdfs_file_with_checkpoint(self.check_path):
# print("实时消费正常完成,删除对应的检查点文件")
# self.kafka_stream_stop()
# else:
# print(f"spider_type:{self.spider_type}还在爬取中, 继续下一个批次数据消费")
# break
# except Exception as e:
# print(f"判断判断爬虫'{self.spider_type}'是否结束---出现异常, 等待20s", e, traceback.format_exc())
# time.sleep(20)
# self.engine_mysql = DBUtil.get_db_engine(db_type=DbTypes.mysql.name, site_name="us")
# continue
def kafka_consumption_is_finished(self): def kafka_consumption_is_finished(self):
"""判断爬虫是否已结束，若已结束则停止kafka消费并退出程序""" """判断爬虫是否已结束，若已结束则停止kafka消费并退出程序"""
max_retries = 10 # 最大重试次数，避免数据库异常时无限循环 max_retries = 10 # 最大重试次数，避免数据库异常时无限循环
...@@ -459,12 +413,27 @@ class Templates(object): ...@@ -459,12 +413,27 @@ class Templates(object):
# 爬虫已完成，停止消费 # 爬虫已完成，停止消费
print(f"爬虫'{self.spider_type}'已爬取完毕({self.site_name} {self.date_type} {self.date_info}), 退出kafka消费") print(f"爬虫'{self.spider_type}'已爬取完毕({self.site_name} {self.date_type} {self.date_info}), 退出kafka消费")
if self.consumer_type == "latest": if self.consumer_type == "latest":
# if HdfsUtils.delete_hdfs_file_with_checkpoint(self.check_path): self.kafka_stream_stop()
# print("实时消费正常完成, 删除对应的检查点文件") elif spider_state == 10 and spider_is_ready == 'yes':
# 爬虫中断，更新kafka_state=1并结束程序
print(f"爬虫'{self.spider_type}'已中断({self.site_name} {self.date_type} {self.date_info}), 更新kafka_state=1并退出kafka消费")
sql = f"UPDATE selection.workflow_manager SET kafka_state=1, updated_at=CURRENT_TIMESTAMP where workflow_name='月全流程' and site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}' and bg_name='{self.site_name}_all_cal'"
if self.date_type == 'day':
sql = f"UPDATE selection.workflow_manager SET kafka_state=1, updated_at=CURRENT_TIMESTAMP where workflow_name='asin详情-天' and site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"
for retry in range(5):
try:
DBUtil.exec_sql('mysql', 'us', sql)
break
except Exception as e:
print(f"UPDATE workflow_manager kafka_state=1 失败（第{retry + 1}次），等待10s重试", e, traceback.format_exc())
if retry == 4:
print("UPDATE workflow_manager kafka_state=1 重试5次后仍失败，放弃")
else:
time.sleep(10)
if self.consumer_type == "latest":
self.kafka_stream_stop() self.kafka_stream_stop()
else: else:
# 爬虫还在进行中 # 爬虫还在进行中
# print(f"爬虫'{self.spider_type}'还在爬取中(spider_state={spider_state}, status_val={status_val}), 继续消费")
print(f"爬虫'{self.spider_type}'还在爬取中(spider_state={spider_state}, spider_is_ready={spider_is_ready}), 继续消费") print(f"爬虫'{self.spider_type}'还在爬取中(spider_state={spider_state}, spider_is_ready={spider_is_ready}), 继续消费")
break break
...@@ -507,15 +476,6 @@ class Templates(object): ...@@ -507,15 +476,6 @@ class Templates(object):
if self.test_flag == 'normal': if self.test_flag == 'normal':
# 启动守护线程，判断是否抓取结束 # 启动守护线程，判断是否抓取结束
self._start_state_monitor_thread() self._start_state_monitor_thread()
# self.kafka_consumption_is_finished()
# 以下超时轮询逻辑已由 _start_state_monitor_thread() 后台线程替代，注释保留备查
# if not has_data:
# elapsed = time.time() - self.last_data_time
# if elapsed > 30 * 60:
# while True:
# time.sleep(120)
# self.kafka_consumption_is_finished()
def handle_kafka_stream(self, kafka_df, epoch_id): def handle_kafka_stream(self, kafka_df, epoch_id):
pass pass
...@@ -530,7 +490,6 @@ class Templates(object): ...@@ -530,7 +490,6 @@ class Templates(object):
if df.shape[0] == 1: if df.shape[0] == 1:
end_offsets_json = list(df.end_offsets_json)[0] end_offsets_json = list(df.end_offsets_json)[0]
print(f"end_offsets_json: {end_offsets_json}") print(f"end_offsets_json: {end_offsets_json}")
# self.beginning_offsets_dict = json.loads(end_offsets_json) # history消费时, 初始的偏移量
self.beginning_offsets_dict = eval(end_offsets_json) # history消费时, 初始的偏移量 self.beginning_offsets_dict = eval(end_offsets_json) # history消费时, 初始的偏移量
def record_offsets_by_history(self, end_offsets_dict): def record_offsets_by_history(self, end_offsets_dict):
...@@ -600,15 +559,6 @@ class Templates(object): ...@@ -600,15 +559,6 @@ class Templates(object):
else: else:
partition_offsets_dict[key]['beginning_offsets'] = end_offsets partition_offsets_dict[key]['beginning_offsets'] = end_offsets
# 当kafka主题有数据正在生产/数据自动删除时, 就需要及时更新起始偏移量
# 删除时
# if beginning_offsets_dict[str(key)] < partition_offsets_dict_check[key]['beginning_offsets']:
# beginning_offsets_dict[str(key)] = partition_offsets_dict_check[key]['beginning_offsets']
# partition_offsets_dict[key]['beginning_offsets'] = partition_offsets_dict_check[key]['beginning_offsets']
# # 生产时
# if partition_offsets_dict[key]['end_offsets'] < partition_offsets_dict_check[key]['end_offsets']:
# partition_offsets_dict[key]['end_offsets'] = partition_offsets_dict_check[key]['end_offsets']
starting_offsets_json = json.dumps({topic_name: beginning_offsets_dict}) starting_offsets_json = json.dumps({topic_name: beginning_offsets_dict})
ending_offsets_json = json.dumps({topic_name: end_offsets_dict}) ending_offsets_json = json.dumps({topic_name: end_offsets_dict})
print(f"starting_offsets_json: {starting_offsets_json}, ending_offsets_json:{ending_offsets_json}") print(f"starting_offsets_json: {starting_offsets_json}, ending_offsets_json:{ending_offsets_json}")
...@@ -638,34 +588,10 @@ class Templates(object): ...@@ -638,34 +588,10 @@ class Templates(object):
time.sleep(10) time.sleep(10)
continue continue
# break
except Exception as e: except Exception as e:
print(e, traceback.format_exc()) print(e, traceback.format_exc())
time.sleep(10) time.sleep(10)
continue continue
# kafka_df = self.spark.read \
# .format("kafka") \
# .option("kafka.bootstrap.servers", self.kafka_servers) \
# .option("subscribe", topic_name) \
# .option("kafka.security.protocol", self.kafka_security_protocol) \
# .option("kafka.sasl.mechanism", self.kafka_sasl_mechanism) \
# .option("kafka.sasl.jaas.config",
# f'org.apache.kafka.common.security.plain.PlainLoginModule required username="{self.kafka_username}" password="{self.kafka_password}";') \
# .option("failOnDataLoss", "true") \
# .option("startingOffsets", starting_offsets_json) \
# .option("endingOffsets", ending_offsets_json) \
# .load() \
# .select(F.from_json(F.col("value").cast("string"), schema=schema).alias("data")) \
# .select("data.*")
# print(f"kafka_df.count():{kafka_df.count()}")
# print(f"starting_offsets_json: {starting_offsets_json}, ending_offsets_json:{ending_offsets_json}")
#
# if num >= partition_num:
# self.start_process_instance() # 退出之前启动调度
# break
# else:
# self.handle_kafka_history_templates(kafka_df=kafka_df)
# continue
def handle_kafka_history_templates(self, kafka_df): def handle_kafka_history_templates(self, kafka_df):
self.handle_kafka_history(kafka_df) self.handle_kafka_history(kafka_df)
...@@ -733,7 +659,6 @@ class Templates(object): ...@@ -733,7 +659,6 @@ class Templates(object):
CommonUtil.send_wx_msg(wx_users, f"asin详情kafka消费", wx_msg) CommonUtil.send_wx_msg(wx_users, f"asin详情kafka消费", wx_msg)
break break
except Exception as e: except Exception as e:
# print(f"UPDATE workflow_progress 失败（第{retry + 1}次），等待10s重试", e, traceback.format_exc())
print(f"UPDATE workflow_manager 失败（第{retry + 1}次），等待10s重试", e, traceback.format_exc()) print(f"UPDATE workflow_manager 失败（第{retry + 1}次），等待10s重试", e, traceback.format_exc())
if retry == 4: if retry == 4:
CommonUtil.send_wx_msg(wx_users, f"\u26A0asin详情kafka实时消费\u26A0", CommonUtil.send_wx_msg(wx_users, f"\u26A0asin详情kafka实时消费\u26A0",
......
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