# 危險(xiǎn)的寫(xiě)法！
cur.execute("SELECT * FROM huge_table")
rows = cur.fetchall()  # 如果表有1000萬(wàn)行，內(nèi)存直接爆炸
for row in rows:
    process(row)

import psycopg2

def batch_process(conn, batch_size=1000):
    cur = conn.cursor(name='fetch_large_data') # 創(chuàng)建服務(wù)器端游標(biāo)
    cur.execute("SELECT id, raw_data FROM big_table")
    
    results_batch = []
    
    while True:
        # 這里的 fetchmany 是關(guān)鍵，它每次只從服務(wù)器拉取 batch_size 行
        rows = cur.fetchmany(batch_size)
        
        if not rows:
            break # 沒(méi)有數(shù)據(jù)了，徹底跳出循環(huán)
            
        for row in rows:
            # 模擬復(fù)雜的業(yè)務(wù)邏輯處理
            processed_data = heavy_computation(row)
            results_batch.append(processed_data)
            
            # 這里的 break 是一種邏輯控制：
            # 假設(shè)我們?cè)谧鰯?shù)據(jù)清洗，如果發(fā)現(xiàn)某個(gè)標(biāo)志位異常，立即終止本次批次的后續(xù)處理
            if is_corrupted_data(processed_data):
                print("發(fā)現(xiàn)異常數(shù)據(jù)，終止當(dāng)前批次處理")
                break # 跳出內(nèi)層 for 循環(huán)，進(jìn)入下一輪 while 循環(huán)
        
        # 寫(xiě)入數(shù)據(jù)庫(kù)或外部存儲(chǔ)
        write_to_staging_table(results_batch)
        results_batch.clear() # 釋放內(nèi)存
        
    cur.close()

import hashlib

def generate_row_hash(row_data):
    """
    將行數(shù)據(jù)序列化并計(jì)算 MD5 哈希值
    """
    # 假設(shè) row_data 是一個(gè)字典或元組，先轉(zhuǎn)換為標(biāo)準(zhǔn)字符串
    data_str = str(sorted(row_data.items())) if isinstance(row_data, dict) else str(row_data)
    return hashlib.md5(data_str.encode('utf-8')).hexdigest()

def sync_data(conn, source_data_list):
    cur = conn.cursor()
    
    # 1. 獲取目標(biāo)數(shù)據(jù)庫(kù)中已存在的哈希集合
    cur.execute("SELECT row_hash FROM sync_log_table")
    existing_hashes = set([row[0] for row in cur.fetchall()])
    
    insert_list = []
    new_hashes = set()
    
    for item in source_data_list:
        # 2. 計(jì)算當(dāng)前數(shù)據(jù)的哈希
        current_hash = generate_row_hash(item)
        
        # 3. 哈希比對(duì)：如果已存在，跳過(guò)
        if current_hash in existing_hashes or current_hash in new_hashes:
            continue
            
        new_hashes.add(current_hash)
        # 將數(shù)據(jù)和哈希值一起打包準(zhǔn)備插入
        insert_list.append((item['content'], current_hash))
        
        # 4. 利用 break 進(jìn)行內(nèi)存保護(hù)
        # 如果待插入列表過(guò)大，先寫(xiě)入一批，防止內(nèi)存暴漲
        if len(insert_list) >= 5000:
            execute_batch_insert(cur, insert_list)
            conn.commit()
            insert_list.clear()
            
    # 處理剩余數(shù)據(jù)
    if insert_list:
        execute_batch_insert(cur, insert_list)
        conn.commit()
        
    cur.close()

def execute_batch_insert(cursor, data):
    # 使用 psycopg2 的 execute_values 進(jìn)行高效批量插入
    from psycopg2.extras import execute_values
    sql = "INSERT INTO sync_log_table (content, row_hash) VALUES %s"
    execute_values(cursor, sql, data)

def robust_etl_pipeline():
    """
    整合了哈希校驗(yàn)、流式處理（break）和異常重試的完整流程
    """
    conn = get_db_connection()
    # 開(kāi)啟自動(dòng)提交，或者在循環(huán)內(nèi)手動(dòng) commit
    # conn.autocommit = False 
    
    try:
        # 源數(shù)據(jù)游標(biāo)
        source_cur = conn.cursor(name='source_cursor')
        source_cur.execute("SELECT * FROM source_table")
        
        # 目標(biāo)表準(zhǔn)備
        target_cur = conn.cursor()
        
        batch_buffer = []
        processed_count = 0
        
        while True:
            # 1. 流式讀取
            rows = source_cur.fetchmany(1000)
            if not rows:
                print("所有數(shù)據(jù)處理完畢。")
                break
            
            for row in rows:
                # 2. 哈希計(jì)算與校驗(yàn)
                row_hash = hashlib.md5(str(row).encode()).hexdigest()
                
                # 簡(jiǎn)單的去重檢查（實(shí)際中可查表或維護(hù)內(nèi)存集合）
                if is_duplicate(target_cur, row_hash):
                    continue
                
                # 3. 數(shù)據(jù)轉(zhuǎn)換
                transformed = transform(row)
                batch_buffer.append((transformed, row_hash))
                
            # 4. 批量寫(xiě)入
            if batch_buffer:
                try:
                    # 使用 execute_values 高效寫(xiě)入
                    from psycopg2.extras import execute_values
                    execute_values(
                        target_cur,
                        "INSERT INTO target_table (data, hash) VALUES %s ON CONFLICT DO NOTHING",
                        batch_buffer
                    )
                    conn.commit()
                    processed_count += len(batch_buffer)
                    print(f"已處理 {processed_count} 條數(shù)據(jù)...")
                    batch_buffer.clear()
                except Exception as e:
                    print(f"寫(xiě)入失敗: {e}")
                    conn.rollback()
                    # 這里可以加入重試邏輯
                    # time.sleep(5) 并嘗試重連
                    break # 寫(xiě)入失敗，停止處理，防止數(shù)據(jù)污染

    except Exception as e:
        print(f"發(fā)生嚴(yán)重錯(cuò)誤: {e}")
    finally:
        if conn:
            conn.close()

def is_duplicate(cursor, hash_val):
    cursor.execute("SELECT 1 FROM target_table WHERE hash = %s", (hash_val,))
    return cursor.fetchone() is not None

def transform(row):
    # 模擬數(shù)據(jù)轉(zhuǎn)換
    return row[1].upper()