import pandas as pd
import numpy as np

# 創(chuàng)建一個(gè)示例 DataFrame
df = pd.DataFrame({
    '整數(shù)列': np.random.randint(0, 100, 10000),
    '浮點(diǎn)列': np.random.randn(10000),
    '字符串列': np.random.choice(['A', 'B', 'C', 'D', 'E'], 10000),
    '布爾列': np.random.choice([True, False], 10000)
})

# ===== 查看內(nèi)存占用 =====
print("各列內(nèi)存占用：")
print(df.memory_usage(deep=True))
# deep=True 會(huì)精確計(jì)算字符串的內(nèi)存（否則只計(jì)算指針）

print(f"\n總內(nèi)存占用：{df.memory_usage(deep=True).sum() / 1024:.2f} KB")
print(f"總行數(shù)：{len(df)}")

# 不同數(shù)據(jù)類型的內(nèi)存對(duì)比
df_int64 = pd.DataFrame({'A': np.random.randint(0, 100, 10000)})
df_int32 = pd.DataFrame({'A': np.random.randint(0, 100, 10000)}).astype('int32')
df_int8 = pd.DataFrame({'A': np.random.randint(0, 100, 10000)}).astype('int8')
df_float64 = pd.DataFrame({'A': np.random.randn(10000)})
df_float32 = pd.DataFrame({'A': np.random.randn(10000)}).astype('float32')

print(f"int64：{df_int64.memory_usage(deep=True).sum()} bytes")
print(f"int32：{df_int32.memory_usage(deep=True).sum()} bytes")
print(f"int8：{df_int8.memory_usage(deep=True).sum()} bytes")
print(f"float64：{df_float64.memory_usage(deep=True).sum()} bytes")
print(f"float32：{df_float32.memory_usage(deep=True).sum()} bytes")

def optimize_numerical(df):
    """
    自動(dòng)優(yōu)化數(shù)值列的數(shù)據(jù)類型
    """
    df_optimized = df.copy()

    for col in df.select_dtypes(include=['int64']).columns:
        col_min = df[col].min()
        col_max = df[col].max()

        # int8: -128 到 127
        if col_min >= -128 and col_max <= 127:
            df_optimized[col] = df[col].astype('int8')
        # int16: -32768 到 32767
        elif col_min >= -32768 and col_max <= 32767:
            df_optimized[col] = df[col].astype('int16')
        # int32: -2147483648 到 2147483647
        elif col_min >= -2147483648 and col_max <= 2147483647:
            df_optimized[col] = df[col].astype('int32')

    for col in df.select_dtypes(include=['float64']).columns:
        df_optimized[col] = df[col].astype('float32')

    return df_optimized

# 測(cè)試
df = pd.DataFrame({
    '小整數(shù)': np.random.randint(0, 100, 10000),    # 可以用 int8
    '中整數(shù)': np.random.randint(0, 50000, 10000),   # 可以用 int16
    '大整數(shù)': np.random.randint(0, 1000000000, 10000),  # 需要 int32
    '浮點(diǎn)數(shù)': np.random.randn(10000)                # 可以用 float32
})

print("優(yōu)化前：")
print(df.memory_usage(deep=True))

df_opt = optimize_numerical(df)
print("\n優(yōu)化后：")
print(df_opt.memory_usage(deep=True))

reduction = (1 - df_opt.memory_usage(deep=True).sum() / df.memory_usage(deep=True).sum()) * 100
print(f"\n內(nèi)存減少：{reduction:.1f}%")

# 當(dāng)字符串列的**唯一值數(shù)量遠(yuǎn)少于總行數(shù)**時(shí)，使用 category 類型

df = pd.DataFrame({
    '性別': np.random.choice(['男', '女'], 100000),      # 只有 2 個(gè)唯一值
    '城市': np.random.choice(['北京', '上海', '廣州', '深圳'], 100000),  # 4 個(gè)
    '部門': np.random.choice(['技術(shù)', '銷售', '管理', '人事', '財(cái)務(wù)'], 100000),  # 5 個(gè)
    '姓名': [f'用戶{i}' for i in range(100000)]  # 10 萬個(gè)唯一值，不適合 category
})

print("優(yōu)化前：")
for col in df.columns:
    print(f"  {col}: {df[col].memory_usage(deep=True) / 1024:.1f} KB, "
          f"唯一值: {df[col].nunique()}")

# 將低唯一值比例的列轉(zhuǎn)為 category
for col in ['性別', '城市', '部門']:
    df[col] = df[col].astype('category')

print("\n優(yōu)化后：")
for col in df.columns:
    print(f"  {col}: {df[col].memory_usage(deep=True) / 1024:.1f} KB, "
          f"唯一值: {df[col].nunique()}")

total_before = df.memory_usage(deep=True).sum()
df_opt = df.copy()
df_opt['姓名'] = df_opt['姓名'].astype('str')  # 不變
reduction = (1 - df.memory_usage(deep=True).sum() / total_before) * 100
print(f"\n總內(nèi)存減少約：{(1 - df.memory_usage(deep=True).sum() / total_before) * 100:.1f}%")

# ===== 模擬大文件 =====
# 先創(chuàng)建一個(gè)"大" CSV 文件（100 萬行）
np.random.seed(42)
n = 1_000_000
large_df = pd.DataFrame({
    'ID': range(n),
    '類別': np.random.choice(['A', 'B', 'C', 'D', 'E'], n),
    '數(shù)值': np.random.randn(n),
    '金額': np.random.randint(10, 10000, n)
})
# large_df.to_csv('large_data.csv', index=False)  # 實(shí)際文件約 30MB

# ===== 分塊讀取 =====

# 方法一：逐塊處理
chunk_size = 100_000  # 每次讀取 10 萬行
results = []

for chunk in pd.read_csv('large_data.csv', chunksize=chunk_size):
    # 對(duì)每個(gè) chunk 進(jìn)行處理
    # 例如：計(jì)算統(tǒng)計(jì)量
    result = chunk.groupby('類別')['金額'].sum()
    results.append(result)

# 合并所有 chunk 的結(jié)果
final_result = pd.concat(results).groupby(level=0).sum()
print("分塊讀取匯總結(jié)果：")
print(final_result)

# 方法二：只讀取需要的列
chunk = pd.read_csv('large_data.csv', usecols=['類別', '金額'], nrows=1000)
print("\n只讀取指定列：")
print(chunk.head())

# 方法三：在讀取時(shí)指定數(shù)據(jù)類型（省內(nèi)存）
dtypes = {
    'ID': 'int32',
    '類別': 'category',
    '數(shù)值': 'float32',
    '金額': 'int32'
}
chunk = pd.read_csv('large_data.csv', dtype=dtypes, nrows=1000)
print("\n指定數(shù)據(jù)類型讀取：")
print(chunk.memory_usage(deep=True))

def process_large_file(filepath, chunksize=100000):
    """
    分塊處理大文件的典型模式
    """
    # 場(chǎng)景1：逐步累加統(tǒng)計(jì)
    total_count = 0
    total_sum = 0

    for chunk in pd.read_csv(filepath, chunksize=chunksize):
        total_count += len(chunk)
        total_sum += chunk['金額'].sum()

    print(f"總行數(shù)：{total_count}")
    print(f"總金額：{total_sum}")

    # 場(chǎng)景2：逐步構(gòu)建結(jié)果
    results = []
    for chunk in pd.read_csv(filepath, chunksize=chunksize):
        # 對(duì)每個(gè)塊進(jìn)行 groupby 聚合
        agg = chunk.groupby('類別').agg(
            總金額=('金額', 'sum'),
            平均金額=('金額', 'mean'),
            數(shù)量=('金額', 'count')
        )
        results.append(agg)

    # 合并所有塊的結(jié)果
    final = pd.concat(results)
    final = final.groupby(level=0).agg({
        '總金額': 'sum',
        '平均金額': 'mean',
        '數(shù)量': 'sum'
    })
    return final

# ===== 寫入 Parquet =====
# df.to_parquet('data.parquet')

# ===== 讀取 Parquet =====
# df = pd.read_parquet('data.parquet')

# ===== 指定列讀?。ㄖ蛔x需要的列）=====
# df = pd.read_parquet('data.parquet', columns=['類別', '金額'])

# ===== 指定引擎 =====
# pip install pyarrow 或 pip install fastparquet
# df.to_parquet('data.parquet', engine='pyarrow')

import os
import time

# 創(chuàng)建示例數(shù)據(jù)
np.random.seed(42)
n = 500_000
df = pd.DataFrame({
    'ID': range(n),
    '類別': np.random.choice(list('ABCDEFGHIJ'), n),
    '數(shù)值': np.random.randn(n),
    '金額': np.random.randint(10, 10000, n)
})

# 保存為 CSV
csv_path = 'test_data.csv'
df.to_csv(csv_path, index=False)
csv_size = os.path.getsize(csv_path)

# 保存為 Parquet
parquet_path = 'test_data.parquet'
df.to_parquet(parquet_path, index=False)
parquet_size = os.path.getsize(parquet_path)

print(f"CSV 大?。簕csv_size / 1024 / 1024:.2f} MB")
print(f"Parquet 大?。簕parquet_size / 1024 / 1024:.2f} MB")
print(f"壓縮率：{(1 - parquet_size / csv_size) * 100:.1f}%")

# 讀取速度對(duì)比
t0 = time.time()
df_csv = pd.read_csv(csv_path)
t1 = time.time()

t0 = time.time()
df_parquet = pd.read_parquet(parquet_path)
t2 = time.time()

print(f"\nCSV 讀取時(shí)間：{t1 - t0:.3f} 秒")
print(f"Parquet 讀取時(shí)間：{t2 - t0:.3f} 秒")

# 讀取后檢查
print(f"\n數(shù)據(jù)類型保持：")
print(df_parquet.dtypes)

# pip install dask

import dask.dataframe as dd

# ===== 讀取大文件 =====
# dask_df = dd.read_csv('large_data.csv')

# ===== 大部分 Pandas API 在 Dask 中都有 =====
# result = dask_df.groupby('類別')['金額'].sum()

# ===== 觸發(fā)計(jì)算 =====
# result_computed = result.compute()

# ===== Dask 的延遲計(jì)算 =====
# Dask 不會(huì)立即執(zhí)行，而是構(gòu)建一個(gè)"計(jì)算圖"
# 只有調(diào)用 .compute() 時(shí)才會(huì)真正執(zhí)行
# 這使得 Dask 能優(yōu)化執(zhí)行計(jì)劃

# ===== 與 Pandas 的主要區(qū)別 =====
# 1. 不支持 inplace 操作
# 2. 部分 API 不兼容
# 3. 適合"讀取→轉(zhuǎn)換→輸出"的流水線式處理
# 4. 可以跨多臺(tái)機(jī)器運(yùn)行

# pip install memory_profiler

# 在 Jupyter 中使用：
# %load_ext memory_profiler
# %memit df = pd.read_csv('large_data.csv')

import psutil
import os

def get_memory_usage():
    """獲取當(dāng)前進(jìn)程的內(nèi)存使用量"""
    process = psutil.Process(os.getpid())
    mem_info = process.memory_info()
    return mem_info.rss / 1024 / 1024  # MB

print(f"當(dāng)前內(nèi)存使用：{get_memory_usage():.2f} MB")

# 對(duì)比處理前后的內(nèi)存
mem_before = get_memory_usage()
df = pd.DataFrame({
    'A': np.random.randn(1_000_000),
    'B': np.random.choice(list('ABCDE'), 1_000_000)
})
mem_after = get_memory_usage()
print(f"創(chuàng)建 DataFrame 消耗：{mem_after - mem_before:.2f} MB")

# ===== 技巧匯總 =====

# 1. 讀取時(shí)只選擇需要的列
df = pd.read_csv('data.csv', usecols=['列1', '列2', '列3'])

# 2. 在讀取時(shí)指定合適的數(shù)據(jù)類型
df = pd.read_csv('data.csv', dtype={
    'ID': 'int32',
    '類別': 'category',
    '數(shù)值': 'float32'
})

# 3. 用更小的數(shù)值類型
# int64 → int32/int16/int8
# float64 → float32

# 4. 字符串列轉(zhuǎn) category（唯一值少時(shí)）
df['分類'] = df['分類'].astype('category')

# 5. 刪除不需要的列釋放內(nèi)存
# del df['不需要的列']
# 或
df = df.drop(columns=['不需要的列'])

# 6. 及時(shí)釋放不需要的變量
# import gc
# del big_df
# gc.collect()

# 7. 使用 Parquet 格式存儲(chǔ)中間結(jié)果
# df.to_parquet('cleaned.parquet')

# 8. 大文件分塊讀取
# for chunk in pd.read_csv('big.csv', chunksize=100000):
#     process(chunk)

# 參考答案
import pandas as pd
import numpy as np

# 創(chuàng)建模擬大數(shù)據(jù)集
np.random.seed(42)
n = 500_000
df = pd.DataFrame({
    'ID': range(n),
    '性別': np.random.choice(['男', '女'], n),
    '省份': np.random.choice([
        '北京', '上海', '廣東', '浙江', '江蘇', '山東', '河南', '四川'
    ], n),
    '年齡段': np.random.choice(['18-25', '26-35', '36-45', '46-55', '56+'], n),
    '評(píng)分': np.random.randint(1, 6, n),
    '消費(fèi)金額': np.random.uniform(10, 10000, n).round(2),
    '訂單數(shù)': np.random.randint(0, 100, n),
    '標(biāo)簽': np.random.choice(['A', 'B', 'C'], n),
    '備注': [f'備注{i}' for i in range(n)]
})

# 原始內(nèi)存
original_mem = df.memory_usage(deep=True).sum()
print(f"原始內(nèi)存：{original_mem / 1024 / 1024:.2f} MB")

# ===== 優(yōu)化 =====
df_opt = df.copy()

# 1. 數(shù)值列降級(jí)
df_opt['ID'] = df_opt['ID'].astype('int32')
df_opt['評(píng)分'] = df_opt['評(píng)分'].astype('int8')
df_opt['訂單數(shù)'] = df_opt['訂單數(shù)'].astype('int16')
df_opt['消費(fèi)金額'] = df_opt['消費(fèi)金額'].astype('float32')

# 2. 字符串列轉(zhuǎn) category
for col in ['性別', '省份', '年齡段', '標(biāo)簽']:
    df_opt[col] = df_opt[col].astype('category')

optimized_mem = df_opt.memory_usage(deep=True).sum()
print(f"優(yōu)化后內(nèi)存：{optimized_mem / 1024 / 1024:.2f} MB")

reduction = (1 - optimized_mem / original_mem) * 100
print(f"內(nèi)存減少：{reduction:.1f}%")

# 驗(yàn)證數(shù)據(jù)一致性
assert (df['ID'] == df_opt['ID']).all()
assert (df['消費(fèi)金額'].round(2) == df_opt['消費(fèi)金額'].round(2)).all()
print("數(shù)據(jù)一致性驗(yàn)證通過！")

# 參考答案
import pandas as pd
import numpy as np
import time
import os

# 創(chuàng)建測(cè)試數(shù)據(jù)
np.random.seed(42)
n = 300_000
df = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=n, freq='H'),
    'category': np.random.choice(list('ABCDEFGH'), n),
    'value': np.random.randn(n),
    'amount': np.random.randint(1, 1000, n)
})

# 寫入測(cè)試
t0 = time.time()
df.to_csv('test.csv', index=False)
csv_write = time.time() - t0

t0 = time.time()
df.to_parquet('test.parquet', index=False)
parquet_write = time.time() - t0

# 讀取測(cè)試
t0 = time.time()
df1 = pd.read_csv('test.csv')
csv_read = time.time() - t0

t0 = time.time()
df2 = pd.read_parquet('test.parquet')
parquet_read = time.time() - t0

# 文件大小
csv_size = os.path.getsize('test.csv')
parquet_size = os.path.getsize('test.parquet')

print("=== 性能對(duì)比 ===")
print(f"寫入時(shí)間 - CSV: {csv_write:.3f}s, Parquet: {parquet_write:.3f}s")
print(f"讀取時(shí)間 - CSV: {csv_read:.3f}s, Parquet: {parquet_read:.3f}s")
print(f"文件大小 - CSV: {csv_size/1024/1024:.2f}MB, Parquet: {parquet_size/1024/1024:.2f}MB")
print(f"壓縮率: {(1 - parquet_size/csv_size)*100:.1f}%")

# 數(shù)據(jù)類型保持
print("\n=== 數(shù)據(jù)類型 ===")
print(f"CSV 讀取后:\n{df1.dtypes}")
print(f"\nParquet 讀取后:\n{df2.dtypes}")