pip install pandas pyspark pyarrow

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

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

import zipfile
import json
import io

def read_json_from_zip_basic(zip_path, json_filename):
    """基礎(chǔ)方法：讀取ZIP中的單個(gè)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

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

def zip_json_to_pandas_simple(zip_path, json_filename):
    """將ZIP中的JSON文件轉(zhuǎn)換為Pandas DataFrame（簡(jiǎn)單版）"""
    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是對(duì)象格式，可能需要特殊處理
    else:
        return pd.DataFrame([json_data])

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

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進(jìn)行流式解析
            parser = ijson.parse(json_file)
            
            for prefix, event, value in parser:
                # 根據(jù)JSON結(jié)構(gòu)進(jìn)行相應(yīng)處理
                if event == 'start_array' or event == 'end_array':
                    continue
                # 這里需要根據(jù)實(shí)際JSON結(jié)構(gòu)調(diào)整解析邏輯
                
    return pd.DataFrame(items)

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

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)

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()

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

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

from concurrent.futures import ThreadPoolExecutor

def parallel_zip_processing(zip_paths, processing_function, max_workers=4):
    """并行處理多個(gè)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"""
        
        # 如果未指定文件名，自動(dòng)查找第一個(gè)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"""
        
        # 使用臨時(shí)目錄解壓文件
        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)換浮點(diǎn)類型
                df[col] = pd.to_numeric(df[col], downcast='float')
        
        return df
    
    def close(self):
        """關(guān)閉Spark會(huì)話"""
        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()