Python生成器遍历案例如何读取数据

wen python案例 2026-06-30 26

本文目录导读：

Python生成器遍历案例如何读取数据

基础文件读取 - 逐行读取大文件
数据分块读取 - 处理CSV文件
数据库查询结果流式处理
实时数据流处理
多层嵌套数据处理
实用数据处理管道

我来为您介绍Python生成器遍历数据的几个典型案例,从基础到实用场景。

基础文件读取 - 逐行读取大文件

def read_large_file(file_path):
    """逐行读取大文件，避免内存溢出"""
    with open(file_path, 'r', encoding='utf-8') as file:
        for line in file:
            # 去除换行符并yield返回
            yield line.strip()
# 使用示例
def process_large_file():
    file_gen = read_large_file('large_data.txt')
    # 方式1：for循环遍历
    for line in file_gen:
        if line:  # 跳过空行
            print(f"处理数据: {line[:50]}...")  # 只显示前50字符
            # 进行数据处理
    # 方式2：手动控制迭代
    file_gen = read_large_file('large_data.txt')
    first_10_lines = [next(file_gen) for _ in range(10)]
    print(f"前10行数据: {first_10_lines}")

数据分块读取 - 处理CSV文件

import csv
def chunked_csv_reader(file_path, chunk_size=1000):
    """分块读取CSV文件，每次返回指定行数"""
    with open(file_path, 'r', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        chunk = []
        for row in reader:
            chunk.append(row)
            if len(chunk) == chunk_size:
                yield chunk
                chunk = []
        # 返回最后不足chunk_size的数据
        if chunk:
            yield chunk
# 使用示例
def process_csv_in_chunks():
    csv_gen = chunked_csv_reader('data.csv', chunk_size=500)
    for chunk_num, chunk_data in enumerate(csv_gen, 1):
        print(f"处理第 {chunk_num} 批数据，共 {len(chunk_data)} 行")
        # 批量处理数据
        for row in chunk_data:
            # 处理每一行数据
            processed = {
                'name': row['name'].strip(),
                'age': int(row['age']),
                'city': row['city'].upper()
            }
            # 这里可以写入数据库或做其他处理

数据库查询结果流式处理

import sqlite3
def stream_database_records(db_path, table_name, batch_size=100):
    """流式读取数据库记录，避免一次性加载所有数据"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    offset = 0
    while True:
        cursor.execute(
            f"SELECT * FROM {table_name} LIMIT ? OFFSET ?",
            (batch_size, offset)
        )
        records = cursor.fetchall()
        if not records:
            break
        for record in records:
            yield record
        offset += batch_size
    conn.close()
# 使用示例
def process_database_records():
    records_gen = stream_database_records('database.db', 'users')
    for record in records_gen:
        user_id, name, email = record
        print(f"处理用户: {name} ({email})")
        # 处理每条记录

实时数据流处理

import time
import random
def real_time_data_stream():
    """模拟实时数据流"""
    sensors = ['temperature', 'humidity', 'pressure']
    while True:
        data = {
            'timestamp': time.time(),
            'sensor': random.choice(sensors),
            'value': random.uniform(20, 30),
            'unit': 'celsius' if random.choice(sensors) == 'temperature' else 'percent'
        }
        yield data
        time.sleep(0.1)  # 模拟数据采集间隔
# 使用示例
def monitor_sensor_data():
    data_stream = real_time_data_stream()
    # 收集最近10秒的数据
    start_time = time.time()
    collected_data = []
    for data in data_stream:
        collected_data.append(data)
        # 每收集100条数据做一次分析
        if len(collected_data) >= 100:
            analyze_batch(collected_data)
            collected_data = []
        # 检查是否超过10秒
        if time.time() - start_time > 10:
            print("数据采集完成")
            break
def analyze_batch(data_batch):
    """分析一批数据"""
    temperatures = [d['value'] for d in data_batch if d['sensor'] == 'temperature']
    if temperatures:
        avg_temp = sum(temperatures) / len(temperatures)
        print(f"平均温度: {avg_temp:.2f}°C")

多层嵌套数据处理

def flatten_nested_json(data, parent_key='', sep='_'):
    """扁平化嵌套JSON数据"""
    if isinstance(data, dict):
        for key, value in data.items():
            new_key = f"{parent_key}{sep}{key}" if parent_key else key
            yield from flatten_nested_json(value, new_key, sep)
    elif isinstance(data, list):
        for i, item in enumerate(data):
            new_key = f"{parent_key}{sep}{i}" if parent_key else str(i)
            yield from flatten_nested_json(item, new_key, sep)
    else:
        yield parent_key, data
# 使用示例
def process_nested_data():
    nested_data = {
        'user': {
            'name': 'Alice',
            'contacts': [
                {'type': 'email', 'value': 'alice@example.com'},
                {'type': 'phone', 'value': '123-456-7890'}
            ],
            'settings': {
                'theme': 'dark',
                'notifications': True
            }
        }
    }
    # 扁平化处理
    for flat_key, value in flatten_nested_json(nested_data):
        print(f"{flat_key}: {value}")

实用数据处理管道

def data_pipeline(data_source):
    """数据管道示例：清洗 -> 转换 -> 验证 -> 输出"""
    def clean_data(stream):
        """数据清洗"""
        for item in stream:
            item = item.strip()
            if item:  # 跳过空数据
                yield item
    def transform_data(stream):
        """数据转换"""
        for item in stream:
            # 转换为大写并添加前缀
            yield f"DATA_{item.upper()}"
    def validate_data(stream):
        """数据验证"""
        valid_prefix = "DATA_"
        for item in stream:
            if item.startswith(valid_prefix) and len(item) > 5:
                yield item
    # 组合管道
    cleaned = clean_data(data_source)
    transformed = transform_data(cleaned)
    validated = validate_data(transformed)
    return validated
# 使用示例
def use_data_pipeline():
    raw_data = ["  apple  ", "banana", "", "  cherry  ", "date", None, "elderberry"]
    # 过滤None值并创建生成器
    clean_source = (item for item in raw_data if item is not None)
    # 应用数据管道
    pipeline = data_pipeline(clean_source)
    print("处理后的数据:")
    for processed_item in pipeline:
        print(f"  - {processed_item}")

内存优化：使用生成器替代列表，处理大数据时特别有效
流式处理：按需生成数据，而不是一次性加载全部
管道组合：多个生成器可以串联形成数据处理管道
延迟计算：数据只有在需要时才被处理
终止控制：使用break或条件判断来控制无限生成器

这些案例覆盖了常见的生成器应用场景,您可以根据实际需求调整和使用。