deepwalk refactor for memsave and perfermance optimize

tangwang
1 parent 06cb25fa
Showing 12 changed files with 17 additions and 175 deletions Show diff stats
offline_tasks/config.py
offline_tasks/scripts/db_service.py -> offline_tasks/db_service.py
offline_tasks/scripts/add_names_to_swing.py
offline_tasks/scripts/config.py
offline_tasks/scripts/fetch_item_attributes.py
offline_tasks/scripts/generate_session.py
offline_tasks/scripts/i2i_content_similar.py
offline_tasks/scripts/i2i_deepwalk.py
offline_tasks/scripts/i2i_session_w2v.py
offline_tasks/scripts/i2i_swing.py
offline_tasks/scripts/interest_aggregation.py
offline_tasks/scripts/load_index_to_redis.py
@@ -1,26 +0,0 @@
-import os  # Add for environment variable reading
-
-
-ES_CONFIG = {
-    'host': 'http://localhost:9200',
-    # default index name will be overwritten below based on APP_ENV
-    'index_name': 'spu',
-    'username': 'essa',
-    'password': '4hOaLaf41y2VuI8y'
-}
-
-
-# Redis Cache Configuration
-REDIS_CONFIG = {
-    # 'host': '120.76.41.98',
-    'host': 'localhost',
-    'port': 6479,
-    'snapshot_db': 0,
-    'password': 'BMfv5aI31kgHWtlx',
-    'socket_timeout': 1,
-    'socket_connect_timeout': 1,
-    'retry_on_timeout': False,
-    'cache_expire_days': 180,  # 6 months
-    'translation_cache_expire_days': 360,
-    'translation_cache_prefix': 'trans'
-}
@@ -5,7 +5,7 @@
 """
 import argparse
 from datetime import datetime
-from debug_utils import setup_debug_logger, load_name_mappings_from_file
+from scripts.debug_utils import setup_debug_logger, load_name_mappings_from_file
  
  
 def add_names_to_swing_result(input_file, output_file, name_mappings, logger=None, debug=False):
@@ -1,130 +0,0 @@
-"""
-离线任务配置文件
-包含数据库连接、路径、参数等配置
-"""
-import os
-from datetime import datetime, timedelta
-
-# 数据库配置
-DB_CONFIG = {
-    'host': 'selectdb-cn-wuf3vsokg05-public.selectdbfe.rds.aliyuncs.com',
-    'port': '9030',
-    'database': 'datacenter',
-    'username': 'readonly',
-    'password': 'essa1234'
-}
-
-# 路径配置
-BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-OUTPUT_DIR = os.path.join(BASE_DIR, 'output')
-LOG_DIR = os.path.join(BASE_DIR, 'logs')
-
-# 确保目录存在
-os.makedirs(OUTPUT_DIR, exist_ok=True)
-os.makedirs(LOG_DIR, exist_ok=True)
-
-# ============================================================================
-# 默认参数配置（用于调试和生产）
-# ============================================================================
-
-# 时间配置（建议先用小数值调试，确认无误后再改为大数值）
-DEFAULT_LOOKBACK_DAYS = 400    # 默认回看天数（调试用30天，生产可改为730天）
-DEFAULT_RECENT_DAYS = 180       # 默认最近天数（调试用7天，生产可改为180天）
-
-# i2i算法默认参数
-DEFAULT_I2I_TOP_N = 50        # 默认返回Top N个相似商品
-
-# 兴趣聚合默认参数
-DEFAULT_INTEREST_TOP_N = 1000  # 默认每个key返回Top N个商品
-
-# 获取时间范围
-def get_time_range(days=DEFAULT_LOOKBACK_DAYS):
-    """获取时间范围"""
-    end_date = datetime.now()
-    start_date = end_date - timedelta(days=days)
-    return start_date.strftime('%Y-%m-%d'), end_date.strftime('%Y-%m-%d')
-
-# i2i 行为相似算法配置
-I2I_CONFIG = {
-    # Swing 算法配置
-    'swing': {
-        'alpha': 0.5,           # swing算法的alpha参数
-        'threshold1': 0.5,      # 交互强度阈值1
-        'threshold2': 0.5,      # 交互强度阈值2
-        'max_sim_list_len': 300,  # 最大相似列表长度
-        'top_n': 50,            # 输出top N个相似商品
-        'thread_num': 10,       # 线程数（如果使用C++版本）
-    },
-    
-    # Session W2V 配置
-    'session_w2v': {
-        'max_sentence_length': 100,  # 最大句子长度
-        'window_size': 5,            # 窗口大小
-        'vector_size': 128,          # 向量维度
-        'min_count': 2,              # 最小词频
-        'workers': 10,               # 训练线程数
-        'epochs': 10,                # 训练轮数
-        'sg': 1,                     # 使用skip-gram
-    },
-    
-    # DeepWalk 配置
-    'deepwalk': {
-        'num_walks': 10,         # 每个节点的游走次数
-        'walk_length': 40,       # 游走长度
-        'window_size': 5,        # 窗口大小
-        'vector_size': 128,      # 向量维度
-        'min_count': 2,          # 最小词频
-        'workers': 10,           # 训练线程数
-        'epochs': 10,            # 训练轮数
-        'sg': 1,                 # 使用skip-gram
-        'use_softmax': True,     # 使用softmax
-        'temperature': 1.0,      # softmax温度
-        'p_tag_walk': 0.2,       # 通过标签游走的概率
-    }
-}
-
-# 兴趣点聚合配置
-INTEREST_AGGREGATION_CONFIG = {
-    'top_n': 1000,              # 每个key生成前N个商品
-    'time_decay_factor': 0.95,  # 时间衰减因子（每30天）
-    'min_interaction_count': 2,  # 最小交互次数
-    
-    # 行为权重
-    'behavior_weights': {
-        'click': 1.0,
-        'addToCart': 3.0,
-        'addToPool': 2.0,
-        'contactFactory': 5.0,
-        'purchase': 10.0,
-    },
-    
-    # 类型配置
-    'list_types': ['hot', 'cart', 'new'],  # 热门、加购、新品
-}
-
-# Redis配置（用于存储索引）
-REDIS_CONFIG = {
-    'host': 'localhost',
-    'port': 6379,
-    'db': 0,
-    'password': None,
-    'decode_responses': False
-}
-
-# 日志配置
-LOG_CONFIG = {
-    'level': 'INFO',
-    'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-    'date_format': '%Y-%m-%d %H:%M:%S'
-}
-
-# Debug配置
-DEBUG_CONFIG = {
-    'enabled': False,           # 是否开启debug模式
-    'log_level': 'DEBUG',       # debug日志级别
-    'sample_size': 5,           # 数据采样大小
-    'save_readable': True,      # 是否保存可读明文文件
-    'log_dataframe_info': True, # 是否记录DataFrame详细信息
-    'log_intermediate': True,   # 是否记录中间结果
-}
-
@@ -8,8 +8,8 @@ import json
 import argparse
 from datetime import datetime
 from db_service import create_db_connection
-from config import DB_CONFIG, OUTPUT_DIR
-from debug_utils import setup_debug_logger
+from config.offline_config import DB_CONFIG, OUTPUT_DIR
+from scripts.debug_utils import setup_debug_logger
  
  
 def fetch_and_save_mappings(engine, output_dir, logger=None, debug=False):
@@ -9,11 +9,11 @@ from collections import defaultdict
 import argparse
 from datetime import datetime, timedelta
 from db_service import create_db_connection
-from config import (
+from config.offline_config import (
     DB_CONFIG, OUTPUT_DIR, get_time_range,
     DEFAULT_LOOKBACK_DAYS
 )
-from debug_utils import setup_debug_logger, log_dataframe_info
+from scripts.debug_utils import setup_debug_logger, log_dataframe_info
  
  
 def aggregate_user_sessions(df, behavior_weights, logger=None, debug=False):
@@ -9,8 +9,8 @@ import pandas as pd
 from datetime import datetime, timedelta
 from elasticsearch import Elasticsearch
 from db_service import create_db_connection
-from config import DB_CONFIG, OUTPUT_DIR
-from debug_utils import setup_debug_logger, log_processing_step
+from config.offline_config import DB_CONFIG, OUTPUT_DIR
+from scripts.debug_utils import setup_debug_logger, log_processing_step
  
 # ES配置
 ES_CONFIG = {
@@ -6,24 +6,22 @@ i2i - DeepWalk算法实现
 import pandas as pd
 import argparse
 import os
-import sys
 from datetime import datetime
 from collections import defaultdict
 from gensim.models import Word2Vec
 from db_service import create_db_connection
-from config import (
+from config.offline_config import (
     DB_CONFIG, OUTPUT_DIR, I2I_CONFIG, get_time_range,
     DEFAULT_LOOKBACK_DAYS, DEFAULT_I2I_TOP_N
 )
-from debug_utils import (
+from scripts.debug_utils import (
     setup_debug_logger, log_dataframe_info,
     save_readable_index, fetch_name_mappings, log_algorithm_params,
     log_processing_step
 )
  
 # 导入 DeepWalk 实现
-sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(__file__)), 'deepwalk'))
-from deepwalk import DeepWalk
+from deepwalk.deepwalk import DeepWalk
  
  
 def build_edge_file_from_db(df, behavior_weights, output_path, logger):
@@ -10,11 +10,11 @@ from collections import defaultdict
 from gensim.models import Word2Vec
 import numpy as np
 from db_service import create_db_connection
-from config import (
+from config.offline_config import (
     DB_CONFIG, OUTPUT_DIR, I2I_CONFIG, get_time_range,
     DEFAULT_LOOKBACK_DAYS, DEFAULT_I2I_TOP_N
 )
-from debug_utils import (
+from scripts.debug_utils import (
     setup_debug_logger, log_dataframe_info, log_dict_stats,
     save_readable_index, fetch_name_mappings, log_algorithm_params,
     log_processing_step
@@ -10,11 +10,11 @@ import argparse
 import json
 from datetime import datetime, timedelta
 from db_service import create_db_connection
-from config import (
+from config.offline_config import (
     DB_CONFIG, OUTPUT_DIR, I2I_CONFIG, get_time_range,
     DEFAULT_LOOKBACK_DAYS, DEFAULT_I2I_TOP_N
 )
-from debug_utils import (
+from scripts.debug_utils import (
     setup_debug_logger, log_dataframe_info, log_dict_stats,
     save_readable_index, load_name_mappings_from_file, log_algorithm_params,
     log_processing_step
@@ -9,11 +9,11 @@ import json
 from datetime import datetime, timedelta
 from collections import defaultdict, Counter
 from db_service import create_db_connection
-from config import (
+from config.offline_config import (
     DB_CONFIG, OUTPUT_DIR, INTEREST_AGGREGATION_CONFIG, get_time_range,
     DEFAULT_LOOKBACK_DAYS, DEFAULT_RECENT_DAYS, DEFAULT_INTEREST_TOP_N
 )
-from debug_utils import (
+from scripts.debug_utils import (
     setup_debug_logger, log_dataframe_info, log_dict_stats,
     save_readable_index, fetch_name_mappings, log_algorithm_params,
     log_processing_step
@@ -6,7 +6,7 @@ import redis
 import argparse
 import logging
 from datetime import datetime
-from config import REDIS_CONFIG, OUTPUT_DIR
+from config.offline_config import REDIS_CONFIG, OUTPUT_DIR
  
 logging.basicConfig(
     level=logging.INFO,