快捷導(dǎo)航

pandas實(shí)現(xiàn)處理TB級(jí)別的數(shù)據(jù)

更新時(shí)間：2025年04月06日 11:35:49 作者：若木胡

這篇文章主要介紹了pandas實(shí)現(xiàn)處理TB級(jí)別的數(shù)據(jù),具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

分塊讀取數(shù)據(jù)

Pandas 提供了 chunksize 參數(shù)，它允許你分塊讀取大型文件，每次處理一部分?jǐn)?shù)據(jù)，這樣可以避免內(nèi)存不足的問(wèn)題。

以 CSV 文件為例：

import pandas as pd

# 定義每次讀取的行數(shù)
chunk_size = 100000
for chunk in pd.read_csv('large_file.csv', chunksize=chunk_size):
    # 在這里對(duì)每個(gè)數(shù)據(jù)塊進(jìn)行處理
    processed_chunk = chunk[chunk['column_name'] > 10]
    # 可以將處理后的數(shù)據(jù)塊保存或進(jìn)一步聚合
    # 例如，將處理后的數(shù)據(jù)塊追加到一個(gè)文件中
    processed_chunk.to_csv('processed_file.csv', mode='a', header=not bool(chunk_number))
    chunk_number += 1

在上述代碼里，借助 chunksize 參數(shù)對(duì) CSV 文件進(jìn)行分塊讀取，每次讀取指定行數(shù)的數(shù)據(jù)，對(duì)每個(gè)數(shù)據(jù)塊進(jìn)行處理之后，再把處理后的數(shù)據(jù)塊追加到新文件里。

使用 Dask 與 Pandas 結(jié)合

Dask 是一個(gè)靈活的并行計(jì)算庫(kù)，它可以處理比內(nèi)存更大的數(shù)據(jù)集。

Dask 的 DataFrame API 與 Pandas 類似，這使得你可以使用熟悉的 Pandas 操作來(lái)處理大型數(shù)據(jù)集。

import dask.dataframe as dd

# 讀取大型 CSV 文件
df = dd.read_csv('large_file.csv')

# 進(jìn)行一些數(shù)據(jù)處理操作
result = df[df['column_name'] > 10].groupby('group_column').sum()

# 計(jì)算結(jié)果
final_result = result.compute()

上述代碼利用 Dask 的 read_csv 函數(shù)讀取大型 CSV 文件，構(gòu)建一個(gè) Dask DataFrame，接著進(jìn)行數(shù)據(jù)處理操作，最后調(diào)用 compute 方法計(jì)算最終結(jié)果。

數(shù)據(jù)壓縮和類型優(yōu)化

在讀取數(shù)據(jù)時(shí)，對(duì)數(shù)據(jù)類型進(jìn)行優(yōu)化，采用合適的數(shù)據(jù)類型來(lái)減少內(nèi)存占用。

例如，使用 astype 方法把整數(shù)列的數(shù)據(jù)類型從 int64 轉(zhuǎn)換為 int32 或 int16。

import pandas as pd

# 讀取數(shù)據(jù)
df = pd.read_csv('large_file.csv')

# 優(yōu)化數(shù)據(jù)類型
df['integer_column'] = df['integer_column'].astype('int32')
df['float_column'] = df['float_column'].astype('float32')

代碼中使用 astype 方法對(duì)整數(shù)列和浮點(diǎn)數(shù)列的數(shù)據(jù)類型進(jìn)行優(yōu)化，從而減少內(nèi)存占用。

數(shù)據(jù)庫(kù)查詢和篩選

如果數(shù)據(jù)存儲(chǔ)在數(shù)據(jù)庫(kù)中，可以利用數(shù)據(jù)庫(kù)的查詢功能，只選擇需要的列和行，避免將整個(gè)數(shù)據(jù)集加載到內(nèi)存中。

import pandas as pd
import sqlite3

# 連接到數(shù)據(jù)庫(kù)
conn = sqlite3.connect('large_database.db')

# 執(zhí)行查詢語(yǔ)句，只選擇需要的數(shù)據(jù)
query = "SELECT column1, column2 FROM large_table WHERE condition = 'value'"
df = pd.read_sql(query, conn)

# 關(guān)閉數(shù)據(jù)庫(kù)連接
conn.close()

上述代碼通過(guò) sqlite3 連接到數(shù)據(jù)庫(kù)，執(zhí)行查詢語(yǔ)句，只選擇需要的列和行，然后將查詢結(jié)果加載到 Pandas DataFrame 中。

分布式計(jì)算

對(duì)于超大規(guī)模的數(shù)據(jù)，可以考慮使用分布式計(jì)算框架，如 Apache Spark。

Spark 能夠處理 PB 級(jí)別的數(shù)據(jù)，并且提供了與 Pandas 類似的 API（PySpark Pandas API），方便進(jìn)行數(shù)據(jù)處理。

import pyspark.pandas as ps

# 創(chuàng)建 SparkSession
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("LargeDataProcessing").getOrCreate()

# 讀取大型 CSV 文件
df = ps.read_csv('large_file.csv')

# 進(jìn)行數(shù)據(jù)處理操作
result = df[df['column_name'] > 10].groupby('group_column').sum()

# 輸出結(jié)果
print(result)

# 停止 SparkSession
spark.stop()

此代碼使用 PySpark Pandas API 讀取大型 CSV 文件，進(jìn)行數(shù)據(jù)處理操作，最后輸出結(jié)果。