分布式数据库（GBase8a数据库）随机采样方法（oracle与mysql的Bernoulli采样方法平替）

1   主要解决问题

（1）在GBase8a数据库缺少随机采样方法。

（2）由于分布式数据库的特点，各节点层rowid有重复，单纯依赖rowid无法做到随机采样。

 2   使用方式

（1）修改脚本参数

    DB_CONFIG = {
        'host': '192.168.1.5',         ---集群ip
        'user': 'gbase',               ---用户名  
        'password': 'gbase20110531',   ---密码
        'n': 0.02                      ---采样率，范围从0.000001到100   
    }

（2）直接运行脚本： python gbase_sample.py

3.  验证效果

当采样率设置为0.01时：

python a.py
表名：test.lineorder，数据量：3000000
采样率：0.010000，采集样本数量：29965

当采样率设置为0.02时：
python a.py
表名：test.lineorder，数据量：3000000
采样率：0.020000，采集样本数量：59936

4. 参考文件

'''
作者: Bryan Zhao
创建时间: 2025-07-17
'''

import pymysql
import random

def generate_random_numbers(n):
    """
    生成0-65535之间的随机数集合m，使得m中元素的个数与65536的比为n。
    
    参数:
    n (float): 目标比值，范围从0.000001到100
    
    返回:
    str: 包含随机选择的数字的字符串，数字之间用逗号分隔
    """
    # 验证输入n的范围
    if n < 0.000001 or n > 100:
        raise ValueError("n必须在0.000001到100之间")
    
    # 计算需要选择的数字个数
    count = int(65536 * n)
    
    # 确保至少选择一个数字，并处理n极大的情况
    count = max(1, min(count, 65536))  # 限制范围在1到65536之间
    
    # 随机选择指定数量的唯一数字（范围是0到65535，共65536个数字）
    numbers = random.sample(range(65536), count)
    
    # 将列表转换为用逗号分隔的字符串
    return ','.join(map(str, numbers))

def Bernoulli_sample(host, user, password, n, db='test', table='lineorder', port=5258):
    """
    连接Gbase数据库并对指定表进行Bernoulli采样
    
    参数:
    host (str): 数据库主机名
    user (str): 数据库用户名
    password (str): 数据库密码
    n (float): 采样率，范围从0.000001到100
    db (str): 数据库名，默认为'test'
    table (str): 表名，默认为'lineorder'
    port (int): 数据库端口，默认为5258
    
    返回:
    dict: 包含采样结果的统计信息
    """
    # 生成随机数集合
    random_numbers = generate_random_numbers(n)
    
    # 构建SQL查询
    full_table_name = f"{db}.{table}"
    sql = f"SELECT COUNT(*) FROM {full_table_name} WHERE rowid%65536 IN ({random_numbers})"
    
    try:
        # 连接数据库
        conn = pymysql.connect(
            host=host,
            user=user,
            password=password,
            port=port,
            database=db
        )
        
        with conn.cursor() as cursor:
            # 获取表的总行数
            count_sql = f"SELECT COUNT(*) FROM {full_table_name}"
            cursor.execute(count_sql)
            total_rows = cursor.fetchone()[0]
            
            # 执行采样查询
            cursor.execute(sql)
            sample_count = cursor.fetchone()[0]
            
        # 关闭数据库连接
        conn.close()
        
        # 返回结果报告
        return {
            'table_name': full_table_name,
            'total_rows': total_rows,
            'sampling_rate': n,
            'sample_count': sample_count
        }
        
    except Exception as e:
        print(f"数据库操作错误: {e}")
        return None

if __name__ == "__main__":
    # 数据库连接参数（请根据实际情况修改）
    DB_CONFIG = {
        'host': '192.168.1.5',
        'user': 'gbase',
        'password': 'gbase20110531',
        'n': 0.02  # 采样率，例如0.1%
    }
    
    # 执行采样并获取结果
    result = Bernoulli_sample(**DB_CONFIG)
    
    # 输出报告
    if result:
        print(f"表名：{result['table_name']}，数据量：{result['total_rows']}")
        print(f"采样率：{result['sampling_rate']:.6f}，采集样本数量：{result['sample_count']}")
    else:
        print("采样过程中发生错误，未能获取结果。")