新增redis-存储报错-消息推送

Pyspark_job/my_kafka/kafka_asin_detail.py

@@ -53,7 +53,7 @@ class DimStAsinInfo(Templates):
             "fr": 4,
             "it": 5,
         }
-        self.client = redis.Redis(host='192.168.10.224', port=6379, db=self.redis_db[self.site_name], password='yswg2023')
+        self.client = redis.Redis(host='192.168.10.224', port=6379, db=self.redis_db[self.site_name], password='N8#rTp2Xz!Lk6@Vw9qHs4&Yb1Fm0Cj3')
         self.db_save = f'kafka_asin_detail'
         self.app_name = self.get_app_name()
         self.spark = self.create_spark_object(
@@ -535,35 +535,45 @@ class DimStAsinInfo(Templates):
         # 将Spark DataFrame转换为Pandas DataFrame
         pdf = df.toPandas()
         print(f"开始存储数据: {pdf.shape}")
-        # 遍历Pandas DataFrame并将数据插入到Redis
-        for index, row in pdf.iterrows():
-            # 创建一个复合键，或者根据你的需要选择适当的键
-            # 1. 外层key为10197， 内层可以为10197:15931
-            # redis_key = f"{row['key_outer']}:{row['key_inner']}"  #
-            # # 插入值到Redis - 在这里我仅仅存储了一个值，你可以存储一个字典来存储多个值
-            # self.client.set(redis_key, row['value'])
-            # row_json = row.to_json(orient='split')
-            # self.client.set(redis_key, row_json)
-            # 2. 外层key为10197， 内层可以为15931
-            # redis_key = row['key_outer']
-            # redis_field = row['key_inner']
-            # row_json = row.to_json(orient='split')
-            # self.client.hset(redis_key, redis_field, row_json)
-
-            # 3. hashmap + 外层key为10197， 内层可以为15931
-            redis_key = row['key_outer']
-            # redis_field = row['key_inner']
-            redis_field = row['asin']
-            row_dict = row.to_dict()
-            # row_dict = {k: str(v).lower().replace("none", "").replace("nan", "") for k, v in row_dict.items()}  # 确保所有的值都是字符串
-            row_dict = {k: str(v).replace("None", "").replace("none", "").replace("NaN", "").replace("nan", "") for k, v in row_dict.items()}  # 确保所有的值都是字符串
-            row_dict = {k: format(v, ".2f") if isinstance(v, (int, float)) else str(v).replace("None", "").replace(
-                    "nan", "") for k, v in row_dict.items()}
-
-            del row_dict["key_outer"]
-            del row_dict["key_inner"]
-            row_json = json.dumps(row_dict)
-            self.client.hset(redis_key, redis_field, row_json)
+        while True:
+            try:
+                # 遍历Pandas DataFrame并将数据插入到Redis
+                for index, row in pdf.iterrows():
+                    # 创建一个复合键，或者根据你的需要选择适当的键
+                    # 1. 外层key为10197， 内层可以为10197:15931
+                    # redis_key = f"{row['key_outer']}:{row['key_inner']}"  #
+                    # # 插入值到Redis - 在这里我仅仅存储了一个值，你可以存储一个字典来存储多个值
+                    # self.client.set(redis_key, row['value'])
+                    # row_json = row.to_json(orient='split')
+                    # self.client.set(redis_key, row_json)
+                    # 2. 外层key为10197， 内层可以为15931
+                    # redis_key = row['key_outer']
+                    # redis_field = row['key_inner']
+                    # row_json = row.to_json(orient='split')
+                    # self.client.hset(redis_key, redis_field, row_json)
+
+                    # 3. hashmap + 外层key为10197， 内层可以为15931
+                    redis_key = row['key_outer']
+                    # redis_field = row['key_inner']
+                    redis_field = row['asin']
+                    row_dict = row.to_dict()
+                    # row_dict = {k: str(v).lower().replace("none", "").replace("nan", "") for k, v in row_dict.items()}  # 确保所有的值都是字符串
+                    row_dict = {k: str(v).replace("None", "").replace("none", "").replace("NaN", "").replace("nan", "") for k, v in row_dict.items()}  # 确保所有的值都是字符串
+                    row_dict = {k: format(v, ".2f") if isinstance(v, (int, float)) else str(v).replace("None", "").replace(
+                            "nan", "") for k, v in row_dict.items()}
+
+                    del row_dict["key_outer"]
+                    del row_dict["key_inner"]
+                    row_json = json.dumps(row_dict)
+                    self.client.hset(redis_key, redis_field, row_json)
+                break
+            except Exception as e:
+                print(f"存储redis报错--数据库账号密码不对: {e, traceback.format_exc()}")
+                time.sleep(10)
+                self.client = redis.Redis(host='192.168.10.224', port=6379, db=self.redis_db[self.site_name],
+                                          password='N8#rTp2Xz!Lk6@Vw9qHs4&Yb1Fm0Cj3')
+
+                continue
 
 
 if __name__ == '__main__':