啟研學社由知名學者擔任學術顧問，由高校師生與企研數據科學團隊聯合組建的，是以大數據資源及相關技術助力中國學術、智庫與行業研究為宗旨的研究組織。團隊當前的主要目標是挖掘行政、經濟與社會大數據資源在經濟學學術、智庫與相關行業研究領域中的應用價值，以學術研究為標準開展大數據治理研究，努力探索大數據分析技術融入中國經濟社會研究的可行進路。
CSV（逗號分隔值文件格式，有時也稱為字符分隔值，因為分隔字符也可以不是逗號）是一種通用的、相對簡單的文件格式，被用戶、商業和科學廣泛應用。CSV最廣泛的應用是在程序（軟體）之間轉移表格數據。
假設有以下場景，張三從Oracle資料庫中導出一張數據表格發送給王五，王五使用Stata軟體對該表格數據進行計量分析。從資料庫中導出的數據可以保存為txt、csv、xls、json等常見的文件格式，但一般不支持保存為Stata的dta格式，所以張三導出資料庫中表格時，選擇了Stata支持的csv格式。
某次計量分析時，王五發現接收到的CSV文件很大，整個CSV文件大小約為16GB，數據量達到1500萬行，變量有20個以上。遺憾的是，王五的計算機只有16GB內存，如果將這個CSV文件完整得導入到Stata中進行處理，可能會發生內存不足的報錯。
Part2處理大CSV文件的思路王五與張三溝通後，設計了如下的解決方案：
依次處理小CSV文件，刪除不需要的數據，縮減文件大小；
王五用上面的思路處理大CSV文件，處理得到的CSV文件大約為1GB，變量8個，數據量為20萬行，已經可以用Stata進行分析。
王五樂於分享，他把第1步和第3步的python代碼分享給大家。
Part3Python代碼python代碼如下（若要測試代碼請準備好python3.7環境並安裝pandas模塊）：
類PyCSV中有兩個主要的方法：
import pandas as pd
import os
import warnings
warnings.filterwarnings('ignore')
class PyCSV:
def merge_csv(self, save_name, file_dir, csv_encoding='utf-8'):
"""
:param save_name: 合併後保存的文件名稱，需要用戶傳入
:param file_dir: 需要合併的csv文件所在文件夾
:param csv_encoding: csv文件編碼, 默認 utf-8
:return: None
"""
# 合併後保存的文件路徑 = 需要合併文件所在文件夾 + 合併後的文件名稱
self.save_path = os.path.join(file_dir, save_name)
self.__check_name()
# 指定編碼
self.encoding = csv_encoding
# 需要合併的csv文件所在文件夾
self.file_dir = file_dir
self.__check_dir_exist(self.file_dir)
# 文件路徑列表
self.file_list = [os.path.join(self.file_dir, i) for i in os.listdir(self.file_dir)]
self.__check_singal_dir(self.file_list)
# 合併到指定文件中
print("開始合併csv文件 ！")
for file in self.file_list:
df = pd.read_csv(file, encoding=self.encoding)
df.to_csv(self.save_path, index=False, quoting=1, header=not os.path.exists(self.save_path), mode='a')
print(f"{file} 已經被合併到 {self.save_path} ！")
print("所有文件已經合併完成 ！")
def split_csv(self, csv_path, save_dir, split_line=100000, csv_encoding='utf-8'):
"""
切分文件並獲取csv文件信息。
:param csv_path: csv文件路徑
:param save_dir: 切分文件的保存路徑
:param split_line: 按照多少行數進行切分，默認為10萬
:param csv_encoding: csv文件的編碼格式
:return: None
"""
# 傳入csv文件路徑和切分後小csv文件的保存路徑
self.csv_path = csv_path
self.save_dir = save_dir
# 檢測csv文件路徑和保存路徑是否符合規範
self.__check_dir_exist(self.save_dir)
self.__check_file_exist(self.csv_path)
# 設置編碼格式
self.encoding = csv_encoding
# 按照split_line行，進行切分
self.split_line = split_line
print("正在切分文件... ")
# 獲取文件大小
self.file_size = round(os.path.getsize(self.csv_path) / 1024 / 1024, 2)
# 獲取數據行數
self.line_numbers = 0
# 切分後文件的後綴
i = 0
# df生成器，每個元素是一個df，df的行數為split_line，默認100000行
df_iter = pd.read_csv(self.csv_path,
chunksize=self.split_line,
encoding=self.encoding)
# 每次生成一個df，直到數據全部取玩
for df in df_iter:
# 後綴從1開始
i += 1
# 統計數據總行數
self.line_numbers += df.shape[0]
# 設置切分後文件的保存路徑
save_filename = os.path.join(self.save_dir, self.filename + "_" + str(i) + self.extension)
# 列印保存信息
print(f"{save_filename} 已經生成！")
# 保存切分後的數
df.to_csv(save_filename, index=False, encoding='utf-8', quoting=1)
# 獲取數據列名
self.column_names = pd.read_csv(self.csv_path, nrows=10).columns.tolist()
print("切分完畢！")
return None
def __check_dir_exist(self, dirpath):
"""
檢驗 save_dir 是否存在，如果不存在則創建該文件夾。
:return: None
"""
if not os.path.exists(dirpath):
raise FileNotFoundError(f'{dirpath} 目錄不存在，請檢查！')
if not os.path.isdir(dirpath):
raise TypeError(f'{dirpath} 目標路徑不是文件夾，請檢查！')
def __check_file_exist(self, csv_path):
"""
檢驗 csv_path 是否是CSV文件。
:return: None
"""
if not os.path.exists(csv_path):
raise FileNotFoundError(f'{csv_path} 文件不存在，請檢查文件路徑！')
if not os.path.isfile(csv_path):
raise TypeError(f'{csv_path} 路徑非文件格式，請檢查！')
# 文件存在路徑
self.file_path_root = os.path.split(csv_path)[0]
# 文件名稱
self.filename = os.path.split(csv_path)[1].replace('.csv', '').replace('.CSV', '')
# 文件後綴
self.extension = os.path.splitext(csv_path)[1]
if self.extension.upper() != '.CSV':
raise TypeError(f'{csv_path} 文件類型錯誤，非CSV文件類型，請檢查！')
def __check_name(self):
"""
檢查文件名稱是否 .csv 結尾
:return:
"""
if not self.save_path.upper().endswith('.CSV'):
raise TypeError('文件名稱設置錯誤')
def __check_singal_dir(self, file_list):
"""
檢查需要被合併的csv文件所在文件夾是否符合要求。
1. 不應該存在除csv文件以外的文件
2. 不應該存在文件夾。
:return:
"""
for file in file_list:
if os.path.isdir(file):
raise EnvironmentError(f'發現文件夾 {file}, 當前文件夾中存其他文件夾，請檢查！')
if not file.upper().endswith('.CSV'):
raise EnvironmentError(f'發現非CSV文件：{file}, 請確保當前文件夾僅存放csv文件！')
if __name__ == '__main__':
# 測試切分
csv_path = r'E:\simple.csv'
save_dir = r'E:\simple_splited_files'
PyCSV().split_csv(csv_path, save_dir, split_line=10000)
# 測試合併
files_dir = r'E:\simple_splited_files'
save_name = r'merge_simple.csv'
PyCSV().merge_csv(save_name, files_dir)
Part4效果展示1拆分csv文件2合併csv文件點擊搜索你感興趣的內容吧
點擊下方「閱讀全文」了解更多