使用Python高效讀取ZIP壓縮文件中的JSON數(shù)據(jù)

更新時間：2025年10月14日 09:00:36 作者：weixin_30777913

本文將詳細介紹如何使用Python快速高效地讀取ZIP壓縮文件中的UTF-8編碼JSON文件,并將其轉(zhuǎn)換為Pandas DataFrame和PySpark DataFrame,感興趣的小伙伴可以了解下

本文將詳細介紹如何使用Python快速高效地讀取ZIP壓縮文件中的UTF-8編碼JSON文件，并將其轉(zhuǎn)換為Pandas DataFrame和PySpark DataFrame。我們將探討多種方法，包括標準庫方法、優(yōu)化技巧以及處理大文件的策略。

準備工作與環(huán)境設(shè)置

在開始之前，確保已安裝必要的Python庫：

pip install pandas pyspark pyarrow

使用標準庫方法讀取ZIP中的JSON

Python的標準庫zipfile提供了處理ZIP文件的基本功能。以下是基礎(chǔ)讀取方法：

import zipfile
import json
import pandas as pd
from pyspark.sql import SparkSession

# 初始化Spark會話
spark = SparkSession.builder \
    .appName("ZIP_JSON_Reader") \
    .getOrCreate()

高效讀取方法

方法1：使用zipfile和io模塊

import zipfile
import json
import io

def read_json_from_zip_basic(zip_path, json_filename):
    """基礎(chǔ)方法：讀取ZIP中的單個JSON文件"""
    with zipfile.ZipFile(zip_path, 'r') as zip_ref:
        with zip_ref.open(json_filename, 'r') as json_file:
            # 讀取并解析JSON內(nèi)容
            json_data = json.loads(json_file.read().decode('utf-8'))
            return json_data

方法2：批量處理ZIP中的多個JSON文件

def read_multiple_json_from_zip(zip_path, file_extension='.json'):
    """讀取ZIP中所有JSON文件"""
    all_data = []
    
    with zipfile.ZipFile(zip_path, 'r') as zip_ref:
        # 獲取所有JSON文件
        json_files = [f for f in zip_ref.namelist() 
                     if f.endswith(file_extension)]
        
        for json_file in json_files:
            with zip_ref.open(json_file, 'r') as file:
                try:
                    json_data = json.loads(file.read().decode('utf-8'))
                    all_data.append(json_data)
                except json.JSONDecodeError as e:
                    print(f"Error reading {json_file}: {e}")
    
    return all_data

轉(zhuǎn)換為Pandas DataFrame

方法1：直接轉(zhuǎn)換

def zip_json_to_pandas_simple(zip_path, json_filename):
    """將ZIP中的JSON文件轉(zhuǎn)換為Pandas DataFrame（簡單版）"""
    json_data = read_json_from_zip_basic(zip_path, json_filename)
    
    # 如果JSON是數(shù)組格式，直接轉(zhuǎn)換為DataFrame
    if isinstance(json_data, list):
        return pd.DataFrame(json_data)
    # 如果JSON是對象格式，可能需要特殊處理
    else:
        return pd.DataFrame([json_data])

方法2：使用pandas直接讀?。ㄍ扑]）

def zip_json_to_pandas_efficient(zip_path, json_filename):
    """高效方法：使用pandas直接讀取ZIP中的JSON文件"""
    with zipfile.ZipFile(zip_path, 'r') as zip_ref:
        with zip_ref.open(json_filename, 'r') as json_file:
            # 使用pandas直接讀取JSON流
            df = pd.read_json(json_file, encoding='utf-8')
            return df

方法3：處理大型JSON文件

import ijson

def read_large_json_from_zip(zip_path, json_filename):
    """使用流式處理讀取大型JSON文件"""
    items = []
    
    with zipfile.ZipFile(zip_path, 'r') as zip_ref:
        with zip_ref.open(json_filename, 'r') as json_file:
            # 使用ijson進行流式解析
            parser = ijson.parse(json_file)
            
            for prefix, event, value in parser:
                # 根據(jù)JSON結(jié)構(gòu)進行相應(yīng)處理
                if event == 'start_array' or event == 'end_array':
                    continue
                # 這里需要根據(jù)實際JSON結(jié)構(gòu)調(diào)整解析邏輯
                
    return pd.DataFrame(items)

轉(zhuǎn)換為PySpark DataFrame

方法1：通過Pandas中轉(zhuǎn)

def zip_json_to_pyspark_via_pandas(zip_path, json_filename):
    """通過Pandas將ZIP中的JSON轉(zhuǎn)換為PySpark DataFrame"""
    # 先讀取為Pandas DataFrame
    pandas_df = zip_json_to_pandas_efficient(zip_path, json_filename)
    
    # 轉(zhuǎn)換為PySpark DataFrame
    spark_df = spark.createDataFrame(pandas_df)
    return spark_df

方法2：直接使用PySpark讀?。ㄐ杞鈮海?/h3>

import tempfile
import os

def zip_json_to_pyspark_direct(zip_path, json_filename):
    """將ZIP文件解壓后使用PySpark直接讀取"""
    with tempfile.TemporaryDirectory() as temp_dir:
        # 解壓ZIP文件
        with zipfile.ZipFile(zip_path, 'r') as zip_ref:
            zip_ref.extract(json_filename, temp_dir)
        
        # 使用PySpark讀取解壓后的JSON文件
        json_path = os.path.join(temp_dir, json_filename)
        spark_df = spark.read \
            .option("encoding", "UTF-8") \
            .json(json_path)
        
        return spark_df

方法3：處理ZIP中的多個JSON文件

def multiple_zip_json_to_pyspark(zip_path):
    """讀取ZIP中所有JSON文件到PySpark DataFrame"""
    all_dfs = []
    
    with tempfile.TemporaryDirectory() as temp_dir:
        with zipfile.ZipFile(zip_path, 'r') as zip_ref:
            # 解壓所有JSON文件
            json_files = [f for f in zip_ref.namelist() if f.endswith('.json')]
            zip_ref.extractall(temp_dir, json_files)
        
        # 讀取所有JSON文件
        for json_file in json_files:
            json_path = os.path.join(temp_dir, json_file)
            df = spark.read.option("encoding", "UTF-8").json(json_path)
            all_dfs.append(df)
    
    # 合并所有DataFrame
    if all_dfs:
        result_df = all_dfs[0]
        for df in all_dfs[1:]:
            result_df = result_df.union(df)
        return result_df
    else:
        return spark.createDataFrame([], schema=None)

處理大型ZIP文件的策略

方法1：分塊讀取

def read_large_zip_json_chunked(zip_path, json_filename, chunk_size=1000):
    """分塊讀取大型ZIP中的JSON文件"""
    chunks = []
    
    with zipfile.ZipFile(zip_path, 'r') as zip_ref:
        with zip_ref.open(json_filename, 'r') as json_file:
            # 使用pandas的分塊讀取功能
            for chunk in pd.read_json(json_file, encoding='utf-8', 
                                    lines=True, chunksize=chunk_size):
                chunks.append(chunk)
    
    # 合并所有塊
    if chunks:
        return pd.concat(chunks, ignore_index=True)
    else:
        return pd.DataFrame()

方法2：使用內(nèi)存映射

def read_zip_json_with_mmap(zip_path, json_filename):
    """使用內(nèi)存映射處理大型ZIP文件"""
    import mmap
    
    with zipfile.ZipFile(zip_path, 'r') as zip_ref:
        # 獲取文件信息
        file_info = zip_ref.getinfo(json_filename)
        
        with zip_ref.open(json_filename, 'r') as json_file:
            # 創(chuàng)建內(nèi)存映射
            with mmap.mmap(json_file.fileno(), 0, access=mmap.ACCESS_READ) as mmapped_file:
                df = pd.read_json(mmapped_file, encoding='utf-8')
                return df

性能優(yōu)化建議

1. 使用適當?shù)臄?shù)據(jù)類型

def optimize_pandas_dataframe(df):
    """優(yōu)化Pandas DataFrame的內(nèi)存使用"""
    # 轉(zhuǎn)換數(shù)據(jù)類型以減少內(nèi)存使用
    for col in df.columns:
        if df[col].dtype == 'object':
            # 嘗試轉(zhuǎn)換為分類類型
            if df[col].nunique() / len(df) < 0.5:
                df[col] = df[col].astype('category')
        
        # 轉(zhuǎn)換數(shù)值類型
        elif df[col].dtype in ['int64', 'float64']:
            df[col] = pd.to_numeric(df[col], downcast='integer')
    
    return df

2. 并行處理

from concurrent.futures import ThreadPoolExecutor

def parallel_zip_processing(zip_paths, processing_function, max_workers=4):
    """并行處理多個ZIP文件"""
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        results = list(executor.map(processing_function, zip_paths))
    return results

完整示例代碼

import zipfile
import json
import pandas as pd
from pyspark.sql import SparkSession
import tempfile
import os

class ZIPJSONReader:
    """ZIP文件中的JSON讀取器"""
    
    def __init__(self):
        self.spark = SparkSession.builder \
            .appName("ZIP_JSON_Reader") \
            .getOrCreate()
    
    def read_to_pandas(self, zip_path, json_filename=None, optimize=True):
        """讀取ZIP中的JSON文件到Pandas DataFrame"""
        
        # 如果未指定文件名，自動查找第一個JSON文件
        if json_filename is None:
            with zipfile.ZipFile(zip_path, 'r') as zip_ref:
                json_files = [f for f in zip_ref.namelist() 
                            if f.endswith('.json')]
                if not json_files:
                    raise ValueError("No JSON files found in ZIP")
                json_filename = json_files[0]
        
        # 讀取JSON文件
        with zipfile.ZipFile(zip_path, 'r') as zip_ref:
            with zip_ref.open(json_filename, 'r') as json_file:
                df = pd.read_json(json_file, encoding='utf-8')
        
        # 優(yōu)化內(nèi)存使用
        if optimize:
            df = self._optimize_dataframe(df)
        
        return df
    
    def read_to_pyspark(self, zip_path, json_filename=None):
        """讀取ZIP中的JSON文件到PySpark DataFrame"""
        
        # 使用臨時目錄解壓文件
        with tempfile.TemporaryDirectory() as temp_dir:
            with zipfile.ZipFile(zip_path, 'r') as zip_ref:
                if json_filename:
                    # 解壓指定文件
                    zip_ref.extract(json_filename, temp_dir)
                    json_path = os.path.join(temp_dir, json_filename)
                else:
                    # 解壓所有JSON文件
                    json_files = [f for f in zip_ref.namelist() 
                                if f.endswith('.json')]
                    if not json_files:
                        raise ValueError("No JSON files found in ZIP")
                    zip_ref.extractall(temp_dir, json_files)
                    json_path = temp_dir
            
            # 使用PySpark讀取
            df = self.spark.read \
                .option("encoding", "UTF-8") \
                .json(json_path)
            
            return df
    
    def _optimize_dataframe(self, df):
        """優(yōu)化DataFrame內(nèi)存使用"""
        for col in df.columns:
            col_type = df[col].dtype
            
            if col_type == 'object':
                # 轉(zhuǎn)換為分類類型
                num_unique_values = len(df[col].unique())
                num_total_values = len(df[col])
                if num_unique_values / num_total_values < 0.5:
                    df[col] = df[col].astype('category')
            elif col_type in ['int64']:
                # 下轉(zhuǎn)換整數(shù)類型
                df[col] = pd.to_numeric(df[col], downcast='integer')
            elif col_type in ['float64']:
                # 下轉(zhuǎn)換浮點類型
                df[col] = pd.to_numeric(df[col], downcast='float')
        
        return df
    
    def close(self):
        """關(guān)閉Spark會話"""
        self.spark.stop()

# 使用示例
if __name__ == "__main__":
    reader = ZIPJSONReader()
    
    try:
        # 讀取到Pandas
        pandas_df = reader.read_to_pandas('data.zip', 'example.json')
        print("Pandas DataFrame:")
        print(pandas_df.head())
        print(f"Pandas DataFrame shape: {pandas_df.shape}")
        
        # 讀取到PySpark
        spark_df = reader.read_to_pyspark('data.zip', 'example.json')
        print("\nPySpark DataFrame:")
        spark_df.show(5)
        print(f"PySpark DataFrame count: {spark_df.count()}")
        
    finally:
        reader.close()