spider_asin_image.py

import json
import os
import re
import sys
import time
import logging
import traceback
import zlib
import pandas as pd
import redis
from datetime import datetime
sys.path.append("/opt/module/spark-3.2.0-bin-hadoop3.2/demo/py_demo/")
sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from sqlalchemy import create_engine
from utils.templates import Templates
# from ..utils.templates import Templates
from utils.templates_mysql import TemplatesMysql
# from ..utils.templates_mysql import TemplatesMysql
from pyspark.sql.types import IntegerType
from pyspark.sql import functions as F
from pyspark.sql.types import *
from utils.mysql_db import sql_connect, sql_update_many, sql_delete, get_country_engine
from pyspark.sql import SparkSession


class SpiderAsinImg(Templates):

    def __init__(self, site_name='us', consumer_type='lastest', batch_size=100000):
        super(SpiderAsinImg, self).__init__()
        self.site_name = site_name

        self.consumer_type = consumer_type  # 消费实时还是消费历史
        # 通过date_type 获取 topic
        self.get_topic_name()
        # 通过date_type 获取 schema
        self.init_schema()
        # self.topic_name = topic_name  # 主题名字
        self.batch_size = batch_size
        self.batch_size_history = int(batch_size / 10)
        # self.db_save = f'kafka_test_001'
        self.spark = self.create_spark_object(
            app_name=f"{self.db_save}: {self.site_name},{self.date_type}, {self.date_info}")
        # self.schema = self.init_schema()

        # 连接mysql
        self.engine = get_country_engine(self.site_name)
        self.pg14_engine = self.get_14pg_country_engine(self.site_name)
        sql_connect(self.site_name)
        logging.basicConfig(format='%(asctime)s %(name)s %(levelname)s %(message)s',
                            level=logging.INFO)

    def judge_spider_asin_detail_is_finished(self):
        while True:
            try:
                sql = f'SELECT * from workflow_progress WHERE page="ASIN详情" and site_name="{self.site_name}" and date_type="{self.date_type}" and date_info="{self.date_info}" and status_val=3'
                df = pd.read_sql(sql, con=self.engine)
                if df.shape[0] == 1:
                    print(f"ASIN详情状态为3, 抓取完成并终止程序, site_name:{self.site_name}, date_type:{self.date_type}, date_info:{self.date_info}")
                    self.spark.stop()
                    quit()  # 退出程序
                break
            except Exception as e:
                print(e, traceback.format_exc())
                time.sleep(10)
                self.engine = self.get_connection()

    def init_schema(self):
        self.schema = StructType([
            StructField("asin", StringType(), True),
            StructField("img_url", StringType(), True),
            StructField("img_order_by", StringType(), True),
            StructField("data_type", StringType(), True),
            StructField("site", StringType(), True),
        ])
        self.col = ["asin", "img_url", "img_order_by", "data_type", "site"]

    @staticmethod
    def clean_kafka_df(df):
        df = df.withColumnRenamed("seller_id", "account_id")
        # cols_python = ["asin", "parentAsin", "variat_num", "best_sellers_rank", "best_sellers_herf", "price", "rating",
        #         "brand", "brand", "account_id", "account_name", "account_url", "buy_box_seller_type",
        #         "volume", "weight", "weight_str", "launchTime", "total_comments", "page_inventory"]
        # oneCategoryRank, aoVal, bsrOrders, bsrOrdersSale
        # siteName volumeFormat weightFormat asinUpdateTime
        # java那边插件的字段名称
        cols_java = ['asin', 'parentAsin', 'asinVarNum', 'oneCategoryRank', 'bestSellersRank', 'lastHerf', 'aoVal', 'price', 'rating',
                    'bsrOrders', 'bsrOrdersSale', 'brandName', 'accountId', 'accountName', 'accountUrl', 'siteName', 'buyBoxSellerType',
                    'volume', 'volumeFormat', 'weight', 'weightFormat', 'launchTime', 'totalComments', 'pageInventory', 'asinUpdateTime']
        df = df.select("asin", "parentAsin", "variat_num", "best_sellers_rank", "best_sellers_herf", "price", "rating",
                        "brand", "account_id", "account_name", "account_url", "buy_box_seller_type",
                        "volume", "weight", "weight_str", "launch_time", "total_comments", "page_inventory", "asinUpdateTime", "site_name", "node_id")
        return df

    def get_topic_name(self):
        # 需要注意表名问题
        # 月表主题
        self.topic_name = f"{self.site_name}_asin_image"

    def handle_history(self):
        consumer = self.get_kafka_object_by_python(topic_name=self.topic_name)
        partition_data_count = self.get_kafka_partitions_data(consumer=consumer, topic_name=self.topic_name)

        beginning_offsets_list = []
        end_offsets_list = []
        for values in partition_data_count.values():
            beginning_offsets_list.append(values['beginning_offsets'])
            end_offsets_list.append(values['end_offsets'])

        min_offset = min(beginning_offsets_list)
        max_offset = max(end_offsets_list)

        # max_offset = max(partition_data_count.values())
        # for start_offset in range(0, max_offset+1, self.batch_size_history):
        for start_offset in range(min_offset, max_offset+1, self.batch_size_history):
            end_offset = start_offset + self.batch_size_history
            starting_offsets_json = json.dumps({self.topic_name: {str(p): start_offset for p in partition_data_count.keys()}})
            ending_offsets_json = json.dumps({self.topic_name: {str(p): end_offset for p in partition_data_count.keys()}})
            kafka_df = self.spark.read \
                .format("kafka") \
                .option("kafka.bootstrap.servers", self.kafka_servers) \
                .option("subscribe", self.topic_name) \
                .option("startingOffsets", starting_offsets_json) \
                .option("endingOffsets", ending_offsets_json) \
                .option("failOnDataLoss", "false") \
                .load() \
                .select(F.from_json(F.col("value").cast("string"), schema=self.schema).alias("data")) \
                .select("data.*")
            print(f"kafka_df.count():{kafka_df.count()}, start_offset:{start_offset}, end_offset:{end_offset}")
            pdf = kafka_df.toPandas()
            # pdf.to_sql()
            self.data_save(pdf)
        # 关闭SparkSession
        self.spark.stop()

    def get_14pg_country_engine(self, site_name="us"):
        h14_pg_us = {
            "user": "postgres",
            "password": "fazAqRRVV9vDmwDNRNb593ht5TxYVrfTyHJSJ3BS",
            # "host": "61.145.136.61",
            "host": "192.168.10.223",
            "port": "5432",
            # "port": 54328,
            "database": "selection",
        }
        if site_name == 'us' or site_name == 'mx':
            h14_pg_us["database"] = f"selection"
            db_ = 'postgresql+psycopg2://{}:{}@{}:{}/{}'.format(*h14_pg_us.values())
        # elif site_name == "keepa":
        #     db_ = 'mysql+pymysql://{}:{}@{}:{}/{}?charset={}'.format(*h6_pg_us.values())
        else:
            h14_pg_us["database"] = f"selection_{site_name}"
            db_ = 'postgresql+psycopg2://{}:{}@{}:{}/{}'.format(*h14_pg_us.values())
        engine = create_engine(db_, encoding='utf-8')  # , pool_recycle=3600
        return engine

    def data_save(self, df):
        df = df.toPandas()
        # 获取对应表字段
        df = df[self.col]
        if "site" not in df.keys():
            df["site"] = self.site_name
            logging.info("site is not null")
        df["site"] = df['site'].fillna("us")
        # df.drop_duplicates(subset=["asin", "site"], inplace=True)
        for name, group in df.groupby(['site']):
            asins = list(set(group["asin"]))
            logging.info(f"站点{name[0]}  asin：{asins}")
            if name[0] not in ['us', 'de', 'uk', 'it', 'es', 'fr', 'mx', 'ca']:
                logging.info("非8大站点跳过")
                continue
            # if len(asins) == 1:
            #     sql_del = f"delete from {name[0]}_asin_image where asin in ('{tuple(asins)[0]}');"
            # else:
            #     sql_del = f"delete from {name[0]}_asin_image where asin in {tuple(asins)};"
            # self.del_pg_asin(sql_del, site=self.site)
            # logging.info(f"清理{name[0]}_asin_image 表中数据   {asins}")
            del group["site"]
            # group.to_sql(name=f'{name[0]}_asin_image', con=self.pg14_engine, if_exists='append', index=False)
            logging.info(f"入库{name[0]}_asin_image成功")

    def start_stream(self, processing_time=300):
        logging.info(f"主题：{self.topic_name} 每 {processing_time} 秒  存储到数据库")
        kafka_df = self.get_kafka_df_by_spark(schema=self.schema, consumption_type="latest", topics=self.topic_name)
        query = kafka_df.writeStream.foreachBatch(self.process_batch).trigger(processingTime=f'{processing_time} seconds').start()
        query.awaitTermination()

    def process_batch(self, df, epoch_id):
        # self.judge_spider_asin_detail_is_finished()
        logging.info(f"当前批次传输的数据量为df.count():{df.count()}")
        # 确保schema非空以避免NoneType错误
        if not self.schema:
            raise ValueError("Schema is not defined")
        # df.show(5, truncate=False)
        logging.info(f"df.columns:{df.columns}")
        self.data_save(df)
        print("epoch_id:", epoch_id, datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

    def run(self):
        if self.consumer_type == 'lastest':
            self.start_stream(processing_time=10)
        else:
            self.handle_history()


if __name__ == '__main__':
    site_name = sys.argv[1]  # 参数1：站点
    consumer_type = sys.argv[2]  # 参数2：实时 lastest 历史 lastest
    # us day date_info 2023-11-07
    handle_obj = SpiderAsinImg(site_name=site_name, consumer_type=consumer_type, batch_size=100000)
    handle_obj.run()

# /opt/module/spark/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 --master yarn --driver-memory 2g --executor-memory 2g --executor-cores 1 --num-executors 1 --queue spark /opt/module/spark/demo/py_demo/my_kafka/spider_asin_image.py us lastest

# /opt/module/spark/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 --master yarn --driver-memory 2g --executor-memory 2g --executor-cores 1 --num-executors 1 --queue spark /opt/module/spark/demo/py_demo/my_kafka/spider_asin_image.py us lastest

# /opt/module/spark/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 --master yarn --driver-memory 2g --executor-memory 2g --executor-cores 1 --num-executors 1 --queue spark /opt/module/spark/demo/py_demo/my_kafka/spider_asin_image.py us lastest
# for i in `ps -ef|grep "spider_self_asin_detail" |awk '{print $2}' `; do kill -9 $i ; done;