快捷導(dǎo)航

Python如何將大TXT文件分割成4KB小文件

更新時間：2025年04月15日 14:36:04 作者：程序員總部

處理大文本文件是程序員經(jīng)常遇到的挑戰(zhàn),特別是當(dāng)我們需要把一個幾百M(fèi)B甚至幾個GB的TXT文件分割成小塊時,下面我們來聊聊如何用Python自動完成這個任務(wù)吧

處理大文本文件是程序員經(jīng)常遇到的挑戰(zhàn)。特別是當(dāng)我們需要把一個幾百M(fèi)B甚至幾個GB的TXT文件分割成小塊時，手動操作顯然不現(xiàn)實(shí)。今天我們就來聊聊如何用Python自動完成這個任務(wù)，特別是如何精確控制每個分割文件的大小為4KB。

為什么需要分割TXT文件

在實(shí)際開發(fā)中，我們可能會遇到這些情況：

某些老舊系統(tǒng)只能處理小體積文本文件
需要將日志文件分割后上傳到云存儲
進(jìn)行分布式處理時需要將數(shù)據(jù)分片
調(diào)試時需要用小文件快速測試

4KB是個很常用的分割尺寸，因?yàn)樗檬呛芏嘞到y(tǒng)默認(rèn)的內(nèi)存頁大小，處理起來效率很高。那么問題來了：怎么用Python實(shí)現(xiàn)這個需求呢？

基礎(chǔ)版：按行分割

我們先來看一個最簡單的實(shí)現(xiàn)方式：

def split_by_line(input_file, output_prefix, chunk_size=4000):
    with open(input_file, 'r', encoding='utf-8') as f:
        file_count = 1
        current_size = 0
        output_file = None
        
        for line in f:
            line_size = len(line.encode('utf-8'))
            
            if current_size + line_size > chunk_size:
                if output_file:
                    output_file.close()
                output_file = open(f"{output_prefix}_{file_count}.txt", 'w', encoding='utf-8')
                file_count += 1
                current_size = 0
                
            if not output_file:
                output_file = open(f"{output_prefix}_{file_count}.txt", 'w', encoding='utf-8')
                file_count += 1
                
            output_file.write(line)
            current_size += line_size
            
        if output_file:
            output_file.close()

這個腳本可以按行分割文件，盡量保證每個文件不超過指定大小。但是有個問題：它不能精確控制文件大小正好是4KB，特別是當(dāng)某一行特別長時，單個文件可能會超過限制。

進(jìn)階版：精確控制文件大小

要實(shí)現(xiàn)更精確的控制，我們需要按字節(jié)而不是按行來處理：

def split_by_size(input_file, output_prefix, chunk_size=4096):
    with open(input_file, 'rb') as f:
        file_count = 1
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            with open(f"{output_prefix}_{file_count}.txt", 'wb') as out_file:
                out_file.write(chunk)
            file_count += 1

注意！這里我們用二進(jìn)制模式(‘rb’)打開文件，這樣可以精確控制讀取的字節(jié)數(shù)。但是這樣可能會在UTF-8編碼的中文文件中出現(xiàn)亂碼，因?yàn)橐粋€中文字符可能被從中間截?cái)唷?/p>

完美解決方案：支持UTF-8編碼

為了解決中文亂碼問題，我們需要更智能的處理方式：

def split_utf8_safely(input_file, output_prefix, chunk_size=4096):
    buffer = ""
    file_count = 1
    current_size = 0
    
    with open(input_file, 'r', encoding='utf-8') as f:
        while True:
            char = f.read(1)
            if not char:
                if buffer:
                    with open(f"{output_prefix}_{file_count}.txt", 'w', encoding='utf-8') as out_file:
                        out_file.write(buffer)
                break
                
            char_size = len(char.encode('utf-8'))
            if current_size + char_size > chunk_size:
                with open(f"{output_prefix}_{file_count}.txt", 'w', encoding='utf-8') as out_file:
                    out_file.write(buffer)
                file_count += 1
                buffer = ""
                current_size = 0
                
            buffer += char
            current_size += char_size

這個方法逐個字符讀取文件，確保不會截?cái)喽嘧止?jié)字符。雖然速度會慢一些，但能保證分割后的文件都能正常顯示中文內(nèi)容。

性能優(yōu)化：使用緩沖區(qū)

處理大文件時，逐個字符讀取效率太低。我們可以用緩沖區(qū)來提升性能：

def split_with_buffer(input_file, output_prefix, chunk_size=4096, buffer_size=1024):
    buffer = ""
    file_count = 1
    current_size = 0
    
    with open(input_file, 'r', encoding='utf-8') as f:
        while True:
            chunk = f.read(buffer_size)
            if not chunk:
                if buffer:
                    with open(f"{output_prefix}_{file_count}.txt", 'w', encoding='utf-8') as out_file:
                        out_file.write(buffer)
                break
                
            buffer += chunk
            while len(buffer.encode('utf-8')) >= chunk_size:
                # 找到不超過chunk_size的最大子串
                split_pos = 0
                for i in range(1, len(buffer)+1):
                    if len(buffer[:i].encode('utf-8')) <= chunk_size:
                        split_pos = i
                    else:
                        break
                        
                with open(f"{output_prefix}_{file_count}.txt", 'w', encoding='utf-8') as out_file:
                    out_file.write(buffer[:split_pos])
                file_count += 1
                buffer = buffer[split_pos:]
                current_size = 0

處理特殊情況

實(shí)際應(yīng)用中我們還需要考慮一些特殊情況：

文件頭處理：如果需要保留原文件的頭信息到每個分割文件
行完整性：某些場景下需要保持行的完整性
內(nèi)存限制：處理超大文件時的內(nèi)存優(yōu)化
進(jìn)度顯示：添加進(jìn)度條讓長時間運(yùn)行的任務(wù)更友好

這里給出一個保留文件頭的實(shí)現(xiàn)示例：

def split_with_header(input_file, output_prefix, chunk_size=4096, header_lines=1):
    # 先讀取文件頭
    with open(input_file, 'r', encoding='utf-8') as f:
        header = [next(f) for _ in range(header_lines)]
    
    buffer = ""
    file_count = 1
    current_size = len(''.join(header).encode('utf-8'))
    
    with open(input_file, 'r', encoding='utf-8') as f:
        # 跳過已經(jīng)讀取的文件頭
        for _ in range(header_lines):
            next(f)
            
        while True:
            char = f.read(1)
            if not char:
                if buffer:
                    with open(f"{output_prefix}_{file_count}.txt", 'w', encoding='utf-8') as out_file:
                        out_file.writelines(header)
                        out_file.write(buffer)
                break
                
            char_size = len(char.encode('utf-8'))
            if current_size + char_size > chunk_size:
                with open(f"{output_prefix}_{file_count}.txt", 'w', encoding='utf-8') as out_file:
                    out_file.writelines(header)
                    out_file.write(buffer)
                file_count += 1
                buffer = ""
                current_size = len(''.join(header).encode('utf-8'))
                
            buffer += char
            current_size += char_size