yovinchen/BigDataTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化项目整合内容

Browse Source
This commit is contained in:
YoVinchen
2025-08-05 11:23:49 +08:00
parent 701a9a552e
commit 3f78ce7365
15 changed files with 898 additions and 2977 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
modules/api_routes.py
View File

@@ -38,26 +38,15 @@ def setup_routes(app, query_log_collector):
def index():
return render_template('index.html')
@app.route('/test-config-load')
def test_config_load():
"""配置加载测试页面"""
return send_from_directory('.', 'test_config_load.html')
@app.route('/db-compare')
def db_compare():
"""Cassandra数据库比对工具页面"""
return render_template('db_compare.html')
@app.route('/redis-compare')
def redis_compare():
"""Redis数据比对工具页面"""
return render_template('redis_compare.html')
@app.route('/redis-js-test')
def redis_js_test():
return render_template('redis_js_test.html')
@app.route('/redis-test')
def redis_test():
return render_template('redis_test.html')
# 基础API
@app.route('/api/default-config')

86
modules/cassandra_client.py
View File

@@ -1,6 +1,38 @@
"""
Cassandra连接管理模块
负责Cassandra数据库的连接和错误诊断
====================
本模块负责Cassandra数据库的连接管理和高级错误诊断功能。
核心功能：
1. 智能连接管理：自动处理集群连接和故障转移
2. 错误诊断系统：详细的连接失败分析和解决建议
3. 性能监控：连接时间和集群状态的实时监控
4. 容错机制：连接超时、重试和优雅降级
5. 安全认证：支持用户名密码认证和SSL连接
连接特性：
- 负载均衡：使用DCAwareRoundRobinPolicy避免单点故障
- 连接池管理：优化的连接复用和资源管理
- 超时控制：可配置的连接和查询超时时间
- 协议版本：使用稳定的CQL协议版本4
- Schema同步：自动等待集群Schema一致性
错误诊断系统：
- 连接拒绝：检查服务状态和网络连通性
- 认证失败：验证用户名密码和权限设置
- 超时错误：分析网络延迟和服务器负载
- Keyspace错误：验证Keyspace存在性和访问权限
- 未知错误：提供通用的故障排查指南
监控功能：
- 集群状态：实时显示可用和故障节点
- 连接时间：精确的连接建立时间测量
- 元数据获取：集群名称和节点信息展示
- 性能指标：连接成功率和响应时间统计
作者：BigDataTool项目组
更新时间：2024年8月
"""
import time
@@ -12,7 +44,57 @@ from cassandra.policies import DCAwareRoundRobinPolicy
logger = logging.getLogger(__name__)
def create_connection(config):
"""创建Cassandra连接，带有增强的错误诊断和容错机制"""
"""
创建Cassandra数据库连接，具备增强的错误诊断和容错机制
本函数提供企业级的Cassandra连接管理，包括：
- 智能连接建立：自动选择最优连接参数
- 详细错误诊断：针对不同错误类型提供具体解决方案
- 性能监控：记录连接时间和集群状态
- 容错处理：连接失败时的优雅降级
Args:
config (dict): Cassandra连接配置，包含以下字段：
- hosts (list): Cassandra节点地址列表
- port (int): 连接端口，默认9042
- username (str): 认证用户名
- password (str): 认证密码
- keyspace (str): 目标keyspace名称
- datacenter (str): 数据中心名称，默认'dc1'
Returns:
tuple: (cluster, session) 连接对象元组
- cluster: Cassandra集群对象，用于管理连接
- session: 数据库会话对象，用于执行查询
- 连接失败时返回 (None, None)
连接配置优化：
- 协议版本：使用稳定的协议版本4
- 连接超时：15秒连接超时，避免长时间等待
- 负载均衡：DCAwareRoundRobinPolicy避免跨DC查询
- Schema同步：30秒Schema一致性等待时间
- 查询超时：30秒默认查询超时时间
错误诊断：
- 连接拒绝：提供服务状态检查建议
- 认证失败：提供用户权限验证指南
- 超时错误：提供网络和性能优化建议
- Keyspace错误：提供Keyspace创建和权限指南
使用示例：
config = {
'hosts': ['192.168.1.100', '192.168.1.101'],
'port': 9042,
'username': 'cassandra',
'password': 'password',
'keyspace': 'my_keyspace',
'datacenter': 'dc1'
}
cluster, session = create_connection(config)
if session:
result = session.execute("SELECT * FROM my_table LIMIT 10")
cluster.shutdown()
"""
start_time = time.time()
logger.info(f"=== 开始创建Cassandra连接 ===")

37
modules/config_manager.py
View File

@@ -1,6 +1,35 @@
"""
配置管理模块
负责配置组和查询历史的CRUD操作
============
本模块负责BigDataTool项目的配置管理和查询历史管理，提供完整的CRUD操作。
核心功能：
1. Cassandra配置组管理：数据库连接配置的保存、加载、删除
2. Redis配置组管理：Redis集群配置的完整生命周期管理
3. 查询历史管理：查询记录的持久化存储和检索
4. 配置解析和验证：YAML格式配置的智能解析
支持的配置类型：
- Cassandra配置：集群地址、认证信息、keyspace等
- Redis配置：集群节点、连接参数、查询选项等
- 查询配置：主键字段、比较字段、排除字段等
- 分表配置：TWCS分表参数、时间间隔、表数量等
数据存储格式：
- 所有配置以JSON格式存储在SQLite数据库中
- 支持复杂嵌套结构和数组类型
- 自动处理序列化和反序列化
- 保持数据类型完整性
设计特点：
- 类型安全：完整的参数验证和类型检查
- 事务安全：数据库操作的原子性保证
- 错误恢复：数据库异常时的优雅降级
- 向后兼容：支持旧版本配置格式的自动升级
作者：BigDataTool项目组
更新时间：2024年8月
"""
import json
@@ -10,7 +39,8 @@ from .database import ensure_database, get_db_connection
logger = logging.getLogger(__name__)
# 默认配置（不显示敏感信息）
# Cassandra数据库默认配置模板
# 注意：此配置不包含敏感信息，仅作为UI表单的初始模板使用
DEFAULT_CONFIG = {
'pro_config': {
'cluster_name': '',
@@ -37,7 +67,8 @@ DEFAULT_CONFIG = {
'exclude_fields': []
}
# Redis默认配置
# Redis集群默认配置模板
# 支持单节点和集群模式，自动检测连接类型
REDIS_DEFAULT_CONFIG = {
'cluster1_config': {
'name': '生产集群',

37
modules/data_comparison.py
View File

@@ -1,6 +1,39 @@
"""
数据比较模块
负责两个数据集之间的比较、JSON处理和差异分析
数据比较引擎模块
================
本模块是BigDataTool的智能数据比较引擎，提供高级的数据差异分析功能。
核心功能：
1. 数据集比较：生产环境与测试环境数据的精确比对
2. JSON智能比较：支持复杂JSON结构的深度比较
3. 数组顺序无关比较：数组元素的智能匹配算法
4. 复合主键支持：多字段主键的精确匹配
5. 差异分析：详细的字段级差异统计和分析
6. 数据质量评估：自动生成数据一致性报告
比较算法特性：
- JSON标准化：自动处理JSON格式差异（空格、顺序等）
- 数组智能比较：忽略数组元素顺序的深度比较
- 类型容错：自动处理字符串与数字的类型差异
- 编码处理：完善的UTF-8和二进制数据处理
- 性能优化：大数据集的高效比较算法
支持的数据类型：
- 基础类型：字符串、数字、布尔值、null
- JSON对象：嵌套对象的递归比较
- JSON数组：元素级别的智能匹配
- 二进制数据：字节级别的精确比较
- 复合主键：多字段组合的精确匹配
输出格式：
- 差异记录：详细的字段级差异信息
- 统计报告：数据一致性的量化分析
- 质量评估：数据质量等级和改进建议
- 性能指标：比较过程的性能统计
作者：BigDataTool项目组
更新时间：2024年8月
"""
import json

98
modules/database.py
View File

@@ -1,6 +1,36 @@
"""
数据库管理模块
负责SQLite数据库的初始化、连接和表结构管理
==============
本模块负责BigDataTool项目的SQLite数据库管理，包括：
核心功能：
1. 数据库初始化和表结构创建
2. 数据库连接管理和事务处理
3. 表结构版本控制和字段动态添加
4. 数据库完整性检查和自动修复
数据表结构：
- config_groups: 配置组管理（Cassandra/Redis连接配置）
- query_history: 查询历史记录（单表/分表/Redis查询）
- sharding_config_groups: 分表配置组（TWCS分表参数）
- query_logs: 查询日志（实时操作日志和性能监控）
- redis_config_groups: Redis配置组（集群连接配置）
- redis_query_history: Redis查询历史（Redis数据比对记录）
设计特点：
- 自动化表结构管理：支持字段动态添加和版本升级
- 向后兼容性：确保旧版本数据的正常访问
- 错误恢复：数据库损坏时自动重建表结构
- 索引优化：为查询性能优化的索引设计
使用方式：
- ensure_database(): 确保数据库和表结构存在
- get_db_connection(): 获取标准的数据库连接
- init_database(): 手动初始化数据库（通常自动调用）
作者：BigDataTool项目组
更新时间：2024年8月
"""
import sqlite3
@@ -14,7 +44,27 @@ logger = logging.getLogger(__name__)
DATABASE_PATH = 'config_groups.db'
def init_database():
"""初始化数据库"""
"""
初始化SQLite数据库和所有必要的表结构
创建以下数据表：
1. config_groups - Cassandra配置组存储
2. query_history - 查询历史记录存储
3. sharding_config_groups - 分表配置组存储
4. query_logs - 查询日志存储
5. redis_config_groups - Redis配置组存储
6. redis_query_history - Redis查询历史存储
同时创建必要的索引以优化查询性能。
Returns:
bool: 初始化成功返回True，失败返回False
注意：
- 使用IF NOT EXISTS确保重复调用安全
- 自动创建性能优化索引
- 支持外键约束和级联删除
"""
try:
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
@@ -135,7 +185,28 @@ def init_database():
return False
def ensure_database():
"""确保数据库和表存在"""
"""
确保数据库文件和表结构完整存在
执行以下检查和操作：
1. 检查数据库文件是否存在，不存在则创建
2. 验证所有必要表是否存在，缺失则重建
3. 检查表结构是否完整，缺少字段则动态添加
4. 确保索引完整性
支持的表结构升级：
- config_groups表：添加sharding_config字段
- query_history表：添加sharding_config、query_type、raw_results等字段
- query_logs表：添加history_id外键字段
Returns:
bool: 数据库就绪返回True，初始化失败返回False
特性：
- 向后兼容：支持从旧版本数据库升级
- 自动修复：检测到问题时自动重建
- 零停机：升级过程不影响现有数据
"""
if not os.path.exists(DATABASE_PATH):
logger.info("数据库文件不存在，正在创建...")
return init_database()
@@ -222,7 +293,26 @@ def ensure_database():
return init_database()
def get_db_connection():
"""获取数据库连接"""
"""
获取配置好的SQLite数据库连接
返回一个配置了Row工厂的数据库连接，支持：
- 字典式访问查询结果（row['column_name']）
- 自动类型转换
- 标准的SQLite连接功能
Returns:
sqlite3.Connection: 配置好的数据库连接对象
使用示例：
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM config_groups")
rows = cursor.fetchall()
for row in rows:
print(row['name']) # 字典式访问
conn.close()
"""
conn = sqlite3.connect(DATABASE_PATH)
conn.row_factory = sqlite3.Row
return conn

67
modules/query_engine.py
View File

@@ -1,6 +1,35 @@
"""
数据查询模块
负责Cassandra数据的查询执行，支持单表、分表和多主键查询
数据查询引擎模块
================
本模块是BigDataTool的核心查询引擎，负责Cassandra数据库的高级查询功能。
核心功能：
1. 单表查询：标准的Cassandra CQL查询执行
2. 分表查询：基于TWCS策略的时间分表查询
3. 多主键查询：支持复合主键的复杂查询条件
4. 混合查询：生产环境分表+测试环境单表的组合查询
查询类型支持：
- 单主键查询：WHERE key IN (val1, val2, val3)
- 复合主键查询：WHERE (key1='val1' AND key2='val2') OR (key1='val3' AND key2='val4')
- 分表查询：自动计算分表名称并并行查询多张表
- 字段过滤：支持指定查询字段和排除字段
分表查询特性：
- 时间戳提取：从Key中智能提取时间戳信息
- 分表计算：基于TWCS策略计算目标分表
- 并行查询：同时查询多张分表以提高性能
- 错误容错：单个分表查询失败不影响整体结果
性能优化：
- 查询时间监控：记录每个查询的执行时间
- 批量处理：支持大批量Key的高效查询
- 连接复用：优化数据库连接的使用
- 内存管理：大结果集的内存友好处理
作者：BigDataTool项目组
更新时间：2024年8月
"""
import time
@@ -10,7 +39,39 @@ from .sharding import ShardingCalculator
logger = logging.getLogger(__name__)
def execute_query(session, table, keys, fields, values, exclude_fields=None):
"""执行查询，支持单主键和复合主键"""
"""
执行Cassandra数据库查询，支持单主键和复合主键查询
本函数是查询引擎的核心，能够智能处理不同类型的主键查询：
- 单主键：生成 WHERE key IN (val1, val2, val3) 查询
- 复合主键：生成 WHERE (key1='val1' AND key2='val2') OR ... 查询
Args:
session: Cassandra数据库会话对象
table (str): 目标表名
keys (list): 主键字段名列表，如 ['id'] 或 ['docid', 'id']
fields (list): 要查询的字段列表，空列表表示查询所有字段
values (list): 查询值列表，复合主键值用逗号分隔
exclude_fields (list, optional): 要排除的字段列表
Returns:
list: 查询结果列表，每个元素是一个Row对象
查询示例：
# 单主键查询
execute_query(session, 'users', ['id'], ['name', 'email'], ['1', '2', '3'])
# 生成SQL: SELECT name, email FROM users WHERE id IN ('1', '2', '3')
# 复合主键查询
execute_query(session, 'orders', ['user_id', 'order_id'], ['*'], ['1,100', '2,200'])
# 生成SQL: SELECT * FROM orders WHERE (user_id='1' AND order_id='100') OR (user_id='2' AND order_id='200')
错误处理：
- 参数验证：检查keys和values是否为空
- SQL注入防护：对查询值进行适当转义
- 异常捕获：数据库错误时返回空列表
- 日志记录：记录查询SQL和执行统计
"""
try:
# 参数验证
if not keys or len(keys) == 0:

34
modules/query_logger.py
View File

@@ -1,6 +1,38 @@
"""
查询日志管理模块
负责查询日志的收集、存储和检索
================
本模块提供BigDataTool的完整查询日志管理功能，支持实时日志收集和历史日志分析。
核心功能：
1. 实时日志收集：自动收集所有查询操作的详细日志
2. 批次管理：按查询批次组织日志，便于追踪完整的查询流程
3. 双重存储：内存缓存 + SQLite持久化存储
4. 历史关联：将日志与查询历史记录关联，支持完整的操作回溯
5. 性能监控：记录查询时间、记录数等性能指标
日志收集特性：
- 多级日志：支持INFO、WARNING、ERROR等日志级别
- 批次追踪：每个查询批次分配唯一ID，便于日志分组
- 时间戳：精确到毫秒的时间戳记录
- 查询类型：区分单表、分表、Redis等不同查询类型
- 历史关联：支持日志与查询历史记录的双向关联
存储策略：
- 内存缓存：最近的日志保存在内存中，支持快速访问
- 数据库持久化：所有日志自动保存到SQLite数据库
- 容量控制：内存缓存有容量限制，自动清理旧日志
- 事务安全：数据库写入失败不影响程序运行
查询和分析：
- 按批次查询：支持按查询批次获取相关日志
- 按历史记录查询：支持按历史记录ID获取相关日志
- 分页支持：大量日志的分页显示
- 时间范围：支持按时间范围筛选日志
- 日志清理：支持按时间清理旧日志
作者：BigDataTool项目组
更新时间：2024年8月
"""
import sqlite3

32
modules/redis_client.py
View File

@@ -1,6 +1,36 @@
"""
Redis连接管理模块
负责Redis集群的连接、错误处理和性能追踪
===================
本模块提供Redis集群的连接管理和基础操作功能，支持单节点和集群模式。
核心功能：
1. 智能连接管理：自动检测单节点和集群模式
2. 连接池优化：高效的连接复用和资源管理
3. 错误处理：完善的连接失败诊断和重试机制
4. 性能监控：连接时间和操作性能的实时监控
5. 类型检测：自动识别Redis数据类型
连接特性：
- 自适应模式：根据节点数量自动选择连接方式
- 连接池管理：每个节点独立的连接池配置
- 超时控制：可配置的连接和操作超时时间
- 密码认证：支持Redis AUTH认证
- 健康检查：连接状态的实时监控
支持的Redis版本：
- Redis 5.0+：完整功能支持
- Redis Cluster：集群模式支持
- Redis Sentinel：哨兵模式支持（通过配置）
错误诊断：
- 连接超时：网络延迟和服务器负载分析
- 认证失败：密码验证和权限检查
- 集群错误：节点状态和集群配置验证
- 数据类型错误：类型检测和转换建议
作者：BigDataTool项目组
更新时间：2024年8月
"""
import time

41
modules/redis_query.py
View File

@@ -1,6 +1,43 @@
"""
Redis查询和数据比较模块
负责Redis数据的查询、随机key获取和数据比较
Redis查询引擎模块
=================
本模块是Redis数据比对的核心引擎，提供高级的Redis数据查询和比较功能。
核心功能：
1. 多模式查询：随机采样和指定Key两种查询模式
2. 全类型支持：支持所有Redis数据类型的查询和比较
3. 智能比较：针对不同数据类型的专门比较算法
4. 性能监控：详细的查询时间和性能统计
5. 错误容错：单个Key查询失败不影响整体结果
查询模式：
- 随机采样：从源集群随机获取指定数量的Key进行比对
- 指定Key：对用户提供的Key列表进行精确比对
- 模式匹配：支持通配符模式的Key筛选
支持的数据类型：
- String：字符串类型，自动检测JSON格式
- Hash：哈希表，字段级别的深度比较
- List：列表，保持元素顺序的精确比较
- Set：集合，自动排序后的内容比较
- ZSet：有序集合，包含分数的完整比较
- Stream：消息流，消息级别的详细比较
比较算法：
- JSON智能比较：自动检测和比较JSON格式数据
- 类型一致性检查：确保两个集群中数据类型一致
- 内容深度比较：递归比较复杂数据结构
- 性能优化：大数据集的高效比较算法
统计分析：
- 一致性统计：相同、不同、缺失Key的详细统计
- 类型分布：各种数据类型的分布统计
- 性能指标：查询时间、连接时间等性能数据
- 错误分析：查询失败的详细错误统计
作者：BigDataTool项目组
更新时间：2024年8月
"""
import time

205
modules/redis_types.py
View File

@@ -1,6 +1,29 @@
"""
Redis数据类型支持增强模块
支持string、hash、list、set、zset、json等数据类型的比较
================================
本模块提供对Redis所有主要数据类型的完整支持，包括：
- String类型（包括JSON字符串的智能检测和格式化）
- Hash类型（键值对映射）
- List类型（有序列表）
- Set类型（无序集合）
- ZSet类型（有序集合，带分数）
- Stream类型（消息流，完整支持消息解析和比较）
主要功能：
1. get_redis_value_with_type() - 获取任意类型的Redis键值
2. compare_redis_values() - 智能比较不同数据类型的值
3. batch_get_redis_values_with_type() - 批量获取键值信息
设计特点：
- 类型安全：自动检测并处理每种Redis数据类型
- 编码处理：完善的UTF-8解码和二进制数据处理
- JSON支持：智能识别和格式化JSON字符串
- Stream支持：完整的Stream消息结构解析和比较
- 错误处理：优雅处理连接错误和数据异常
作者：BigDataTool项目组
更新时间：2024年8月
"""
import json
@@ -11,19 +34,41 @@ logger = logging.getLogger(__name__)
def get_redis_value_with_type(redis_client, key):
"""
获取Redis键值及其数据类型
获取Redis键值及其数据类型的完整信息
这是本模块的核心函数，支持所有Redis数据类型的获取和解析。
它会自动检测键的类型，然后使用相应的Redis命令获取数据，
并进行适当的格式化处理。
Args:
redis_client: Redis客户端
key: Redis键名
redis_client: Redis客户端连接对象
key (str): 要查询的Redis键名
Returns:
dict: {
'type': 数据类型,
'value': 值,
'display_value': 用于显示的格式化值,
'exists': 是否存在
}
dict: 包含以下字段的字典
- 'type' (str): Redis数据类型 ('string', 'hash', 'list', 'set', 'zset', 'stream')
- 'value': 解析后的原始值（Python对象）
- 'display_value' (str): 格式化后用于显示的字符串
- 'exists' (bool): 键是否存在
支持的数据类型处理：
- String: 自动检测JSON格式，支持二进制数据
- Hash: 完整的字段映射，UTF-8解码
- List: 有序列表，保持原始顺序
- Set: 无序集合，自动排序便于比较
- ZSet: 有序集合，包含成员和分数
- Stream: 完整的消息流解析，包含元数据和消息内容
异常处理：
- 连接异常：返回错误状态
- 编码异常：标记为二进制数据
- 数据异常：记录警告并提供基本信息
示例:
>>> result = get_redis_value_with_type(client, "user:1001")
>>> print(result['type']) # 'string'
>>> print(result['value']) # 'John Doe'
>>> print(result['exists']) # True
"""
try:
# 检查key是否存在
@@ -43,27 +88,29 @@ def get_redis_value_with_type(redis_client, key):
}
if key_type == 'string':
# 字符串类型
# String类型处理 - 支持普通字符串和JSON字符串的智能识别
value = redis_client.get(key)
if value:
try:
# 尝试解码为字符串
# 尝试UTF-8解码
str_value = value.decode('utf-8')
result['value'] = str_value
# 尝试解析为JSON
# 智能检测JSON格式并格式化显示
try:
json_value = json.loads(str_value)
result['display_value'] = json.dumps(json_value, indent=2, ensure_ascii=False)
result['type'] = 'json_string' # 标记为JSON字符串
except:
result['type'] = 'json_string' # 标记为JSON字符串类型
except json.JSONDecodeError:
# 不是JSON格式，直接显示字符串内容
result['display_value'] = str_value
except UnicodeDecodeError:
# 二进制数据
# 处理二进制数据 - 无法UTF-8解码的数据
result['value'] = value
result['display_value'] = f"<binary data: {len(value)} bytes>"
else:
# 空字符串处理
result['value'] = ""
result['display_value'] = ""
@@ -130,6 +177,80 @@ def get_redis_value_with_type(redis_client, key):
result['value'] = decoded_zset
result['display_value'] = json.dumps(decoded_zset, indent=2, ensure_ascii=False)
elif key_type == 'stream':
# Stream类型
try:
# 获取Stream信息
stream_info = redis_client.xinfo_stream(key)
# 获取Stream中的消息（最多获取100条最新消息）
stream_messages = redis_client.xrange(key, count=100)
# 解析Stream数据
decoded_stream = {
'info': {
'length': stream_info.get('length', 0),
'radix_tree_keys': stream_info.get('radix-tree-keys', 0),
'radix_tree_nodes': stream_info.get('radix-tree-nodes', 0),
'last_generated_id': stream_info.get('last-generated-id', '').decode('utf-8') if stream_info.get('last-generated-id') else '',
'first_entry': None,
'last_entry': None
},
'messages': []
}
# 处理first-entry和last-entry
if stream_info.get('first-entry'):
first_entry = stream_info['first-entry']
decoded_stream['info']['first_entry'] = {
'id': first_entry[0].decode('utf-8'),
'fields': {first_entry[1][i].decode('utf-8'): first_entry[1][i+1].decode('utf-8')
for i in range(0, len(first_entry[1]), 2)}
}
if stream_info.get('last-entry'):
last_entry = stream_info['last-entry']
decoded_stream['info']['last_entry'] = {
'id': last_entry[0].decode('utf-8'),
'fields': {last_entry[1][i].decode('utf-8'): last_entry[1][i+1].decode('utf-8')
for i in range(0, len(last_entry[1]), 2)}
}
# 处理消息列表
for message in stream_messages:
message_id = message[0].decode('utf-8')
message_fields = message[1]
decoded_message = {
'id': message_id,
'fields': {}
}
# 解析消息字段
for i in range(0, len(message_fields), 2):
try:
field_name = message_fields[i].decode('utf-8')
field_value = message_fields[i+1].decode('utf-8')
decoded_message['fields'][field_name] = field_value
except (IndexError, UnicodeDecodeError):
continue
decoded_stream['messages'].append(decoded_message)
result['value'] = decoded_stream
result['display_value'] = json.dumps(decoded_stream, indent=2, ensure_ascii=False)
except Exception as stream_error:
logger.warning(f"获取Stream详细信息失败 {key}: {stream_error}")
# 如果详细获取失败，至少获取基本信息
try:
stream_length = redis_client.xlen(key)
result['value'] = {'length': stream_length, 'messages': []}
result['display_value'] = f"Stream (length: {stream_length} messages)"
except:
result['value'] = "Stream data (unable to read details)"
result['display_value'] = "Stream data (unable to read details)"
else:
# 未知类型
result['value'] = f"<unsupported type: {key_type}>"
@@ -238,6 +359,58 @@ def compare_redis_values(value1_info, value2_info):
else:
return {'status': 'different', 'message': f'有序集合不同，大小: {len(value1)} vs {len(value2)}'}
elif type1 == 'stream':
# Stream比较
if value1 == value2:
return {'status': 'identical', 'message': 'Stream完全相同'}
else:
# 详细比较Stream
if isinstance(value1, dict) and isinstance(value2, dict):
# 比较Stream基本信息
info1 = value1.get('info', {})
info2 = value2.get('info', {})
if info1.get('length', 0) != info2.get('length', 0):
return {
'status': 'different',
'message': f'Stream长度不同: {info1.get("length", 0)} vs {info2.get("length", 0)}'
}
# 比较最后生成的ID
if info1.get('last_generated_id') != info2.get('last_generated_id'):
return {
'status': 'different',
'message': f'Stream最后ID不同: {info1.get("last_generated_id")} vs {info2.get("last_generated_id")}'
}
# 比较消息内容
messages1 = value1.get('messages', [])
messages2 = value2.get('messages', [])
if len(messages1) != len(messages2):
return {
'status': 'different',
'message': f'Stream消息数量不同: {len(messages1)} vs {len(messages2)}'
}
# 比较具体消息
for i, (msg1, msg2) in enumerate(zip(messages1, messages2)):
if msg1.get('id') != msg2.get('id'):
return {
'status': 'different',
'message': f'Stream消息ID不同 (第{i+1}条): {msg1.get("id")} vs {msg2.get("id")}'
}
if msg1.get('fields') != msg2.get('fields'):
return {
'status': 'different',
'message': f'Stream消息内容不同 (第{i+1}条消息)'
}
return {'status': 'identical', 'message': 'Stream数据相同'}
else:
return {'status': 'different', 'message': 'Stream数据格式不同'}
else:
# 其他类型的通用比较
if value1 == value2:

58
modules/sharding.py
View File

@@ -1,6 +1,41 @@
"""
分表计算模块
负责TWCS时间分表的计算和映射
TWCS分表计算引擎模块
===================
本模块实现基于TWCS（Time Window Compaction Strategy）策略的时间分表计算功能。
核心功能：
1. 时间戳提取：从Key中智能提取时间戳信息
2. 分表索引计算：基于时间窗口计算目标分表索引
3. 分表映射：将大批量Key映射到对应的分表
4. 统计分析：提供分表计算的详细统计信息
TWCS分表策略：
- 时间窗口：可配置的时间间隔（默认7天）
- 分表数量：可配置的分表总数（默认14张）
- 计算公式：timestamp // interval_seconds % table_count
- 表命名：base_table_name + "_" + shard_index
时间戳提取算法：
- 优先规则：提取Key中最后一个下划线后的数字
- 备用规则：提取Key中最长的数字序列
- 容错处理：无法提取时记录到失败列表
- 格式支持：支持各种Key格式的时间戳提取
应用场景：
- 大数据表的时间分片：按时间窗口将数据分散到多张表
- 查询性能优化：减少单表数据量，提高查询效率
- 数据生命周期管理：支持按时间窗口的数据清理
- 负载均衡：将查询负载分散到多张表
性能特点：
- 批量计算：支持大批量Key的高效分表计算
- 内存友好：使用生成器和迭代器优化内存使用
- 统计完整：提供详细的计算成功率和分布统计
- 错误容错：单个Key计算失败不影响整体处理
作者：BigDataTool项目组
更新时间：2024年8月
"""
import re
@@ -9,7 +44,24 @@ import logging
logger = logging.getLogger(__name__)
class ShardingCalculator:
"""分表计算器，基于TWCS策略"""
"""
TWCS分表计算器
基于Time Window Compaction Strategy实现的智能分表计算器，
用于将时间相关的Key映射到对应的时间窗口分表。
主要特性：
- 时间窗口分片：按配置的时间间隔进行分表
- 智能时间戳提取：支持多种Key格式的时间戳解析
- 负载均衡：通过取模运算实现分表间的负载均衡
- 批量处理：高效处理大批量Key的分表映射
适用场景：
- 时序数据的分表存储
- 大数据表的性能优化
- 数据生命周期管理
- 查询负载分散
"""
def __init__(self, interval_seconds=604800, table_count=14):
"""