快捷導(dǎo)航

pandas與pyspark計(jì)算效率對(duì)比分析

更新時(shí)間：2023年06月16日 09:39:08 作者：一個(gè)散步者的夢(mèng)

這篇文章主要介紹了pandas與pyspark計(jì)算效率對(duì)比,本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下

# 數(shù)據(jù)集構(gòu)造
import pandas as pd
import numpy as np 
import pyarrow
import sys
import time
from pyspark.sql import SparkSession
df = pd.DataFrame(columns=['id','sales'])
df['id']= np.random.randint(1,10,800000000)    
df['sales']= np.random.randint(30,1000,800000000)   # 生成8億數(shù)據(jù)
df = df.append(df)   # 數(shù)據(jù)量膨脹一倍
df.to_parquet('parquet_test')    # 寫入本地文件
print(sys.getsizeof(df) / 1024 / 1024 / 1024)  # 總數(shù)據(jù)占用內(nèi)存：23個(gè)g

定義pandas計(jì)算函數(shù)

pandas的read函數(shù)會(huì)將數(shù)據(jù)一次讀入內(nèi)存，本地機(jī)器資源不夠可能會(huì)有內(nèi)存溢出，這時(shí)候要考慮逐塊讀取，分別對(duì)每塊進(jìn)行聚合，再進(jìn)行累聚合；

def pandas_duration():
    start = time.time()
    # df.to_csv('data.txt',index=False,sep=',')
    df = pd.read_parquet('parquet_test')
    mid_time = time.time()
    print('pandas讀取數(shù)據(jù)用時(shí)：{:.2f}'.format(mid_time-start))
    print(df.groupby('id',as_index=False).max())   # 分組求最大值
    end = time.time()
    print(end-start)

定義pyspark讀取計(jì)算函數(shù)

# 防止driver內(nèi)存溢出，可以把資源調(diào)大點(diǎn)，筆者電腦64個(gè)g就隨意填了個(gè)32g，分區(qū)數(shù)結(jié)合實(shí)際數(shù)據(jù)大小資源調(diào)整
spark = SparkSession.Builder()\
    .master("local[*]")\
    .config("spark.sql.shuffle.partitions",24)\
    .config("spark.driver.memory","32g")\
    .config("spark.driver.maxResultSize","32g")\
    .appName('pyspark')\
    .getOrCreate()
def pyspark_duration():
    start = time.time()
    # df.to_csv('data.txt',index=False,sep=',')
    spark_df = spark.read.parquet('parquet_test')
    mid_time = time.time()
    print('spark讀取數(shù)據(jù)用時(shí)：{:.2f}'.format(mid_time-start))
    spark_df.groupBy('id').agg({"sales":"max"}).show()  # 分組求最大值
    end = time.time()
    print(end-start)

查看spark計(jì)算時(shí)間：

在整個(gè)運(yùn)行過程中，電腦最大內(nèi)存使用14%；（包括其他系統(tǒng)軟件占用），數(shù)據(jù)讀取計(jì)算只花了32秒