diff --git a/load_postgres.py b/load_postgres.py index 9505b21..077c2e3 100644 --- a/load_postgres.py +++ b/load_postgres.py @@ -1,13 +1,11 @@ -import csv -import os import psycopg2 from psycopg2 import sql import multiprocessing as mp from concurrent.futures import ProcessPoolExecutor, as_completed import time -import sys import random import tempfile +import os # ═══════════════════════════════════════════════════════════════ # تنظیمات @@ -20,13 +18,13 @@ DB_CONFIG = { "password": "mypassword" } -CSV_FILE = "numbers.csv" TABLE_NAME = "phone_numbers" COLUMN_NAME = "phone_number" RANDOM_COLUMN_NAME = "random_code" NUM_WORKERS = 12 BATCH_SIZE = 2_000_000 +TOTAL_RECORDS = 125_000_000 # تعداد کل رکوردها - مستقیماً مشخص می‌شود # ═══════════════════════════════════════════════════════════════ # شروع توالی عددی @@ -45,47 +43,16 @@ def optimize_connection(conn): conn.commit() -def get_file_chunks(filename, num_chunks): - """تقسیم فایل به بخش‌های مساوی""" - print(f"Calculating chunks for {filename}...") - file_size = os.path.getsize(filename) - print(f"File size: {file_size / (1024 ** 3):.2f} GB") - chunk_size = file_size // num_chunks - chunks = [] - with open(filename, 'rb') as f: - f.readline() - start = f.tell() - for i in range(num_chunks): - if i < num_chunks - 1: - f.seek(chunk_size * (i + 1)) - f.readline() - end = f.tell() - chunks.append((start, end)) - start = end - else: - chunks.append((start, file_size)) - print(f"Created {len(chunks)} chunks") - return chunks - - -def count_total_rows(): - """شمارش کل رکوردها""" - print("Counting total rows...") - count = 0 - with open(CSV_FILE, 'r') as f: - next(f) - for _ in f: - count += 1 - return count - - -def load_chunk(args): - """هر Worker یک فایل temp می‌سازد و با COPY لود می‌کند""" - chunk_start, chunk_end, worker_id, rows_in_chunk = args - print(f"[Worker {worker_id}] Starting... chunk: {chunk_start} to {chunk_end}") +def load_chunk_sequential(worker_id, start_phone, rows_count): + """ + هر Worker داده‌های توالی خود را تولید کرده و با COPY لود می‌کند + بدون نیاز به هیچ فایل CSV + """ + print(f"[Worker {worker_id}] Starting... Phone range: {start_phone:,} to {start_phone + rows_count - 1:,}") conn = psycopg2.connect(**DB_CONFIG) temp_file = None + try: optimize_connection(conn) print(f"[Worker {worker_id}] Connected to DB") @@ -93,41 +60,27 @@ def load_chunk(args): start_time = time.time() loaded = 0 - # ═══════════════════════════════════════════════════════ - # محاسبه شروع عدد توالی برای این Worker - # ═══════════════════════════════════════════════════════ - phone_start = PHONE_START + (worker_id * rows_in_chunk) - current_phone = phone_start - print(f"[Worker {worker_id}] Phone range: {phone_start:,} to {phone_start + rows_in_chunk - 1:,}") - + # ایجاد فایل موقت temp_fd, temp_file = tempfile.mkstemp(suffix='.csv', prefix=f'worker_{worker_id}_') os.close(temp_fd) + # تولید مستقیم داده‌ها در فایل موقت with open(temp_file, 'w', buffering=8192 * 1024) as temp_f: - with open(CSV_FILE, 'r', buffering=8192 * 1024) as f: - f.seek(chunk_start) + current_phone = start_phone - # رد کردن خط header اگر در ابتدای chunk باشد - if chunk_start > 0: - f.readline() + for i in range(rows_count): + random_code = random.randint(1000, 9999) + temp_f.write(f"{current_phone},{random_code}\n") + current_phone += 1 + loaded += 1 - while f.tell() < chunk_end: - line = f.readline() - if line.strip(): - # ═══════════════════════════════════════ - # استفاده از عدد توالی به جای خواندن از CSV - # ═══════════════════════════════════════ - random_code = random.randint(1000, 9999) - temp_f.write(f"{current_phone},{random_code}\n") - - current_phone += 1 - loaded += 1 - - if loaded % 5_000_000 == 0: - print(f"[Worker {worker_id}] Wrote {loaded:,} rows to temp file") + # نمایش پیشرفت هر worker + if loaded % 5_000_000 == 0: + print(f"[Worker {worker_id}] Generated {loaded:,} rows so far...") print(f"[Worker {worker_id}] Temp file created: {loaded:,} rows") + # لود با COPY with open(temp_file, 'r') as f: with conn.cursor() as cur: cur.copy_expert( @@ -240,10 +193,9 @@ def verify_indexes(): conn.close() -def verify_data(start_check=100_000_000, sample_points=None): +def verify_data(): """بررسی صحت داده‌های لود شده""" - if sample_points is None: - sample_points = [0, 10_000_000, 50_000_000, 100_000_000, 124_999_999] + sample_points = [0, 10_000_000, 50_000_000, 100_000_000, TOTAL_RECORDS - 1] print("\n" + "=" * 50) print("DATA VERIFICATION") @@ -268,17 +220,18 @@ def verify_data(start_check=100_000_000, sample_points=None): def main(): print("=" * 60) - print("PostgreSQL Bulk Loader - Sequential Phone Numbers") + print("PostgreSQL Bulk Loader - Sequential Phone Numbers (No CSV)") print("=" * 60) - print(f"Phone number range: {PHONE_START:,} to {PHONE_START + 125_000_000 - 1:,}") - print(f"Total records: 125,000,000\n") + print(f"Total records: {TOTAL_RECORDS:,}") + print(f"Phone number range: {PHONE_START:,} to {PHONE_START + TOTAL_RECORDS - 1:,}") + print(f"Using {NUM_WORKERS} workers") + print(f"Batch size per worker: {TOTAL_RECORDS // NUM_WORKERS:,}\n") num_workers = min(NUM_WORKERS, mp.cpu_count()) - total_rows = count_total_rows() - print(f"CSV rows: {total_rows:,}") - print(f"Using {num_workers} workers") - print(f"Batch size: {BATCH_SIZE:,}\n") + # محاسبه تعداد رکورد برای هر worker + rows_per_worker = TOTAL_RECORDS // num_workers + remaining_rows = TOTAL_RECORDS % num_workers start_time = time.time() @@ -291,47 +244,41 @@ def main(): conn.close() print("Table ready\n") - # تقسیم فایل - print("Calculating file chunks...") - chunks = get_file_chunks(CSV_FILE, num_workers) + # آماده سازی آرگومان‌ها برای workers + args = [] + current_start = PHONE_START - # ═══════════════════════════════════════════════════════ - # محاسبه تعداد رکوردها در هر chunk - # ═══════════════════════════════════════════════════════ - rows_per_chunk = total_rows // num_workers + for i in range(num_workers): + rows_count = rows_per_worker + (1 if i < remaining_rows else 0) + args.append((i, current_start, rows_count)) + current_start += rows_count - args = [ - (start, end, i, rows_per_chunk) - for i, (start, end) in enumerate(chunks) - ] - - print("\nStarting workers...\n") + print("Starting workers...\n") # اجرای موازی + total_loaded = 0 with ProcessPoolExecutor(max_workers=num_workers) as executor: - futures = [executor.submit(load_chunk, arg) for arg in args] - total = 0 + futures = {executor.submit(load_chunk_sequential, *arg): arg for arg in args} for future in as_completed(futures): try: - total += future.result() + loaded = future.result() + total_loaded += loaded except Exception as e: print(f"Worker failed: {e}") elapsed = time.time() - start_time - rate = total / elapsed if elapsed > 0 else 0 - remaining = (total_rows - total) / rate if rate > 0 else 0 - percent = (total * 100) // total_rows + rate = total_loaded / elapsed if elapsed > 0 else 0 + remaining = (TOTAL_RECORDS - total_loaded) / rate if rate > 0 else 0 + percent = (total_loaded * 100) // TOTAL_RECORDS if TOTAL_RECORDS > 0 else 0 - print(f"\n>>> Progress: {total:,}/{total_rows:,} ({percent}%) | " + print(f"\n>>> Progress: {total_loaded:,}/{TOTAL_RECORDS:,} ({percent}%) | " f"Speed: {rate:,.0f} rows/sec | " f"ETA: {remaining / 60:.1f} min") - print(f"\n✓ Total rows loaded: {total:,}") + print(f"\n✓ Total rows loaded: {total_loaded:,}") - # ═══════════════════════════════════════════════════════ - # ساخت ایندکس‌ها با تأیید - # ═══════════════════════════════════════════════════════ + # ساخت ایندکس‌ها print("\n" + "=" * 60) print("CREATING INDEXES") print("=" * 60) @@ -340,18 +287,14 @@ def main(): success1 = create_index(conn, COLUMN_NAME) conn.close() - # ═══════════════════════════════════════════════════════ # تأیید نهایی - # ═══════════════════════════════════════════════════════ all_ok = verify_indexes() - - # بررسی صحت داده‌ها verify_data() total_time = time.time() - start_time print(f"\n✓ Done! Total time: {total_time / 60:.1f} minutes") - print(f"✓ Average speed: {total_rows / total_time:,.0f} rows/sec") + print(f"✓ Average speed: {TOTAL_RECORDS / total_time:,.0f} rows/sec") if all_ok: print("\n✓✓✓ ALL INDEXES CREATED SUCCESSFULLY ✓✓✓") diff --git a/readme.md b/readme.md new file mode 100644 index 0000000..3d34a29 --- /dev/null +++ b/readme.md @@ -0,0 +1,100 @@ +# Redis vs PostgreSQL Benchmark + +Load 125 million phone number records and benchmark read performance between Redis and PostgreSQL + +## Prerequisites + +pip install psycopg2-binary redis locust + +## PostgreSQL Setup + +sudo -u postgres psql + +CREATE DATABASE numbers_db; +CREATE USER myuser WITH PASSWORD 'mypassword'; +GRANT ALL PRIVILEGES ON DATABASE numbers_db TO myuser; +\q + +## Redis Setup + +sudo systemctl restart redis-server + +## Load Data into PostgreSQL + +python load_postgres.py + +## Load Data into Redis + +python load_redis.py + +## Run Benchmark + +locust -f main.py + +Then open in browser: http://localhost:8089 + +## Default Settings + +Total records: 125,000,000 +Phone number range: 100,000,000 to 224,999,999 +PostgreSQL workers: 12 +Redis workers: 8 + +## Expected Results + +Redis: +- Average latency: 0.5 to 2 ms +- P99 latency: 3 to 8 ms +- Throughput: 40,000 to 60,000 req/s +- Load time: 15 to 20 minutes + +PostgreSQL: +- Average latency: 2 to 10 ms +- P99 latency: 15 to 40 ms +- Throughput: 10,000 to 20,000 req/s +- Load time: 20 to 30 minutes + +## File Structure + +load_postgres.py - PostgreSQL data loader +load_redis.py - Redis data loader +main.py - Locust benchmark script + +## Advanced Settings + +In load_postgres.py: +NUM_WORKERS = 12 +BATCH_SIZE = 2_000_000 + +In load_redis.py: +NUM_WORKERS = 8 +BATCH_SIZE = 100_000 + +## Troubleshooting + +PostgreSQL connection issue: +sudo systemctl status postgresql +sudo ufw allow 5432 + +Redis connection issue: +sudo systemctl status redis-server +redis-cli ping + +Out of memory: +Reduce NUM_WORKERS to 4 in loader files + +## Important Notes + +PostgreSQL table is created as UNLOGGED (faster speed) +Each Locust user has a dedicated database connection +Response time is pure query time (excluding connection) +Phone numbers are sequential starting from 100 million + +## Cleanup + +Remove PostgreSQL database: +DROP DATABASE numbers_db; +DROP USER myuser; + +Clear Redis: +redis-cli FLUSHALL \ No newline at end of file