IterableData

Iterable Data

Iterable Data 是一个 Python 库，可通过一致的迭代器接口逐行读取和写入数据文件。它提供统一的 API，用于处理多种数据格式（CSV、JSON、Parquet、XML 等），类似于 csv.DictReader，但支持更多格式。

该库简化了数据处理和格式转换，同时保留复杂的嵌套数据结构（不同于需要展平数据的 pandas DataFrame）。

功能

• 统一 API：通过单一接口读写多种数据格式
• 自动格式检测：根据文件名或内容（魔数和启发式规则）检测文件类型和压缩格式
• 格式功能报告：以编程方式查询格式功能（读取/写入/批量/总数/流式处理/表格）
• 压缩支持：无缝处理压缩文件
• 保留嵌套数据：将复杂的嵌套结构作为 Python 字典处理
• DuckDB 集成：可选的 DuckDB 引擎，支持通过下推优化实现高性能查询
• 流水线处理：内置流水线支持，可转换数据
• 编码检测：自动检测文本文件的编码和分隔符
• 批量操作：高效地批量读取和写入
• 原生批量转换：可选的列式到列式传输，支持选择列、行范围和批次大小
• 有界列式 I/O：共享行/批量游标和可配置的 Parquet 行组，确保大规模读取和写入不会超出资源限制
• 编解码器性能配置：选择 fast、balanced 或 max 压缩设置，并诊断实际生效的设置
• 表格列表：发现多表格式中可用的表、工作表和数据集
• 上下文管理器支持：使用 with 语句自动清理资源
• DataFrame 桥接：通过单行方法将可迭代数据转换为 Pandas、Polars 和 Dask DataFrame
• 云存储支持：通过 URI 方案直接访问 S3、GCS 和 Azure Blob Storage
• 数据库引擎支持：以可迭代数据源的形式只读访问 SQL 和 NoSQL 数据库（PostgreSQL、ClickHouse、MySQL、MongoDB、Elasticsearch 等）
• 原子写入：使用临时文件和原子重命名，实现适合生产环境的文件写入
• 批量文件转换：使用 glob 模式或目录一次转换多个文件
• 进度跟踪和指标：内置进度条、回调和结构化指标对象
• 错误处理控制：可配置错误策略和结构化错误日志；默认情况下，格式错误的输入会引发带类型的错误，而非被读取为空数据集
• 安全强化：采用 XXE 安全的 XML 解析、基于 AST 白名单的筛选表达式，并明确确认对 pickle 的信任
• 性能回归门禁：CI 对代表性读取/转换工作负载实施基准要求
• 容器格式：从 TAR 归档中流式读取记录，无需将成员文件解压到磁盘
• 类型提示和类型安全：提供完整的类型注解，以及适用于数据类和 Pydantic 模型的带类型辅助函数
• 湖仓表：读取和写入 Delta Lake、Iceberg 和 DuckLake；读取 Hudi；实验性支持 Apache Paimon 表以及 Row/Mosaic 文件格式

支持的文件类型

核心格式

• JSON - 标准 JSON 文件
• JSONL/NDJSON - JSON Lines 格式（每行一个 JSON 对象）
• JSON-LD - 用于关联数据的 JSON（RDF 格式）
• CSV/TSV - 逗号分隔值和制表符分隔值
• 带注释的 CSV - 带有类型注释和元数据的 CSV
• CSVW - Web 上的 CSV（带元数据）
• PSV/SSV - 竖线分隔值和分号分隔值
• LTSV - 带标签的制表符分隔值
• FWF - 固定宽度格式
• XML - 支持可配置标签解析的 XML 文件
• ZIP XML - ZIP 归档中的 XML 文件
• HTML - 支持提取表格的 HTML 文件

二进制格式

• BSON - 二进制 JSON 格式
• MessagePack - 高效的二进制序列化格式
• CBOR - 简明二进制对象表示法
• UBJSON - 通用二进制 JSON
• SMILE - 二进制 JSON 变体
• Bencode - BitTorrent 编码格式
• Avro - Apache Avro 二进制格式（读写）
• Pickle - Python pickle 格式（不可信输入不安全；传入 trust=True 以确认）

列式与分析格式

• Parquet - Apache Parquet 列式格式
• ORC - 优化的列式行格式
• Arrow/Feather - Apache Arrow 列式格式
• GeoParquet - 支持 GeoParquet 元数据的 Parquet 配置文件，可保留几何信息和 CRS
• Lance - 针对 ML 和向量搜索优化的现代列式格式
• Vortex - 支持快速随机访问的现代列式格式
• Paimon Row - Apache Paimon 行格式，支持 O(1) 行号访问
• Paimon Mosaic - 适用于宽表的 Apache Paimon 列式分桶格式
• Paimon - Apache Paimon 数仓/目录表
• Delta Lake - Delta Lake 格式（读写）
• Iceberg - Apache Iceberg 格式（读写）
• DuckLake - DuckLake 湖仓表（读写）
• Hudi - Apache Hudi 格式（读取；暂不支持写入）

数据库格式

• SQLite - SQLite 数据库文件
• DBF - dBase/FoxPro 数据库文件
• MySQL Dump - MySQL 转储文件
• PostgreSQL Copy - PostgreSQL COPY 格式
• DuckDB - DuckDB 数据库文件

统计格式

• SAS - SAS 数据文件
• Stata - Stata 数据文件
• SPSS - SPSS 数据文件
• R Data - R RDS 和 RData 文件
• fst - R fst 列式磁盘帧（fst extra；实验性）
• PX - PC-Axis 格式
• ARFF - 属性-关系文件格式（Weka 格式）
• LIBSVM - 稀疏标记特征向量（读写）
• NumPy - .npy / .npz 数组行（读写；npy extra）

科学格式

• NetCDF - 用于科学数据的网络通用数据格式
• CDF - NASA 通用数据格式（空间科学）
• HDF5 - 分层数据格式
• Zarr - 分块数组存储（zarr extra；实验性）
• XYZ - 分子/点坐标表
• CIF - 晶体学信息文件（atom_site 循环；实验性）
• PDB - 蛋白质数据库 ATOM/HETATM 记录
• MATLAB MAT - MATLAB .mat 变量（mat extra；实验性）
• SEG-Y - 地震波形（geophysical extra；实验性）
• GRIB2 - 气象消息（geophysical extra；实验性）
• miniSEED - 地震波形窗口（geophysical extra；实验性）

地理空间格式

• GeoJSON - 地理 JSON 格式
• GeoJSON Text Sequence - RFC 8142 行分隔的 GeoJSON 要素（.geojsonl、.geojsons）；适合流式处理
• GeoPackage - OGC GeoPackage 格式
• GML - 地理标记语言
• KML - Keyhole 标记语言
• KMZ - 压缩的 KML（包含 KML 的 ZIP 归档）
• GPX - GPS 交换格式（航点、路线、轨迹）
• Shapefile - ESRI Shapefile 格式
• File Geodatabase - 通过 Fiona 读取 ESRI FileGDB 图层（geospatial extra；实验性）
• MapInfo MIF - MapInfo 交换格式（geospatial extra；实验性）
• Esri ASCII Grid - 以单元格或行记录表示的栅格网格（.asc）
• ArcInfo E00 - 交换导出格式（实验性子集）
• LAS - LiDAR 点云（lidar extra；实验性）
• BAG - 测深属性网格（hdf5 extra；实验性）
• CZML - Cesium CZML 文档数据包
• FlatGeobuf - 支持流式处理的地理空间要素，可选空间索引筛选
• MVT/PBF - Mapbox 矢量瓦片
• TopoJSON - 保留拓扑结构的 GeoJSON 扩展

RDF 与语义格式

• JSON-LD - 用于关联数据的 JSON
• RDF/XML - XML 格式的 RDF
• Turtle - 简洁 RDF 三元组语言
• N-Triples - 基于行的 RDF 格式
• N-Quads - 带上下文的 N-Triples
• TriG - RDF 三元组图格式
• N3 - Notation3 RDF 格式
• TriX - XML 三元组 RDF 格式
• HDT - Header-Dictionary-Triples 紧凑 RDF（rdf extra；实验性）

Feed 格式

• Atom - Atom 订阅格式
• RSS - Rich Site Summary feed 格式

网络格式

• PCAP - 数据包捕获格式
• PCAPNG - PCAP Next Generation 格式

日志与事件格式

• Apache Log - Apache 访问/错误日志
• CEF - 通用事件格式
• GELF - Graylog 扩展日志格式
• WARC - Web ARChive 格式
• CDX - Web 归档索引格式
• ILP - InfluxDB 行协议
• HTML - 可提取表格的 HTML 文件

电子邮件格式

• EML - 电子邮件消息格式
• MBOX - 邮箱格式
• MHTML - MIME HTML 格式

配置格式

• INI - INI 配置文件
• TOML - Tom's Obvious Minimal Language
• YAML - YAML Ain't Markup Language
• HOCON - Human-Optimized Config Object Notation
• EDN - 可扩展数据表示法

Office 格式

• XLS/XLSX - Microsoft Excel 文件
• XLSB - Excel 二进制格式
• ODS - OpenDocument 电子表格
• Microsoft Access - Access .mdb / .accdb 表（access extra；实验性）
• Lotus 1-2-3 - 旧版 WK1 / .123 电子表格（实验性）

商务与交换格式

• EDI - X12 / EDIFACT 段流（实验性）
• IATI - 援助透明度活动 XML（xml extra；实验性）

CAD 格式

• DXF - AutoCAD Drawing Exchange Format

图格式

• GraphML - 图标记语言
• GEXF - 图交换 XML 格式
• DOT - GraphViz DOT 格式

序列与比对格式

• FASTA - 序列格式（蛋白质/核苷酸）
• FASTQ - 带质量信息的序列格式
• SAM - Sequence Alignment/Map（文本）
• BAM - 二进制 SAM 格式
• Genomic VCF/BCF - 基因组数据的 Variant Call Format（不同于 vCard .vcf；需要 bio extra）
• CRAM - 参考序列压缩的序列比对（需要 alignment extra，必要时还需显式指定参考序列）
• BED - BED3–BED12 基因组区间
• GFF3/GTF - 基因组特征注释，可保留坐标和属性

流式与大数据格式

• Kafka - Apache Kafka 格式
• Pulsar - Apache Pulsar 格式
• Flink - Apache Flink 格式
• Beam - Apache Beam 格式
• RecordIO - RecordIO 格式
• SequenceFile - Hadoop SequenceFile
• TFRecord - TensorFlow Record 格式
• WebDataset - 将 TAR 分片分组为 ML 样本字典（format="webdataset"；实验性）

协议与序列化格式

• Protocol Buffers - Google Protocol Buffers
• Cap'n Proto - Cap'n Proto 序列化
• FlatBuffers - FlatBuffers 序列化
• FlexBuffers - FlexBuffers 格式
• Thrift - Apache Thrift 格式
• ASN.1 - ASN.1 编码格式
• Ion - Amazon Ion 格式
• OTLP JSON/Protobuf - OpenTelemetry 跟踪、日志和指标导出配置文件（需要 otlp extra）

其他格式

• TAR - 多文件归档容器（只读；以流方式读取成员文件，不解压到磁盘）
• vCard (VCF) - 电子名片（RFC 6350）；不是基因组 Variant Call Format
• iCal - iCalendar 格式
• LDIF - LDAP 数据交换格式
• TXT - 纯文本文件

有关各格式的参数、记录结构和 extras，请参阅 格式文档（或此仓库中的 docs/docs/formats/）。

支持的压缩编解码格式

• GZip (.gz)
• BZip2 (.bz2)
• LZMA (.xz, .lzma)
• LZ4 (.lz4)
• ZIP (.zip)
• Brotli (.br)
• ZStandard (.zst, .zstd)
• Snappy (.snappy, .sz) — 对分帧文件进行流式解压
• LZO (.lzo, .lzop) — 流式解压，并提供旧版 Blob 回退支持
• 7z (.7z；需通过 iterabledata[compression] 安装 py7zr)

要求

Python 3.10+

安装

pip install iterabledata

PyPI 包名为 iterabledata。导入 iterable：

from iterable import open_iterable

也可以从源代码安装：

git clone https://github.com/datenoio/iterabledata.git
cd iterabledata
pip install .

可选依赖项

IterableData 支持可选额外项，以提供更多功能：

# AI-powered documentation generation
pip install iterabledata[ai]

# Database ingestion (PostgreSQL, ClickHouse, MongoDB, MySQL, Elasticsearch, etc.)
pip install iterabledata[db]

# RDF formats (TriG, N3, TriX)
pip install iterabledata[rdf]

# Excel Binary (XLSB)
pip install iterabledata[xlsb]

# Graph formats (GraphML, GEXF, DOT)
pip install iterabledata[graph]

# Alignment formats (BAM, SAM, CRAM)
pip install iterabledata[alignment]

# Genomic formats (VCF/BCF, CRAM, BED, GFF3/GTF via pysam and bio readers)
pip install iterabledata[bio]

# Zarr chunked array stores
pip install iterabledata[zarr]

# GeoParquet, FlatGeobuf, FileGDB, MapInfo MIF
pip install iterabledata[parquet,geospatial]

# LiDAR LAS point clouds
pip install iterabledata[lidar]

# MATLAB .mat files
pip install iterabledata[mat]

# SEG-Y, GRIB2, miniSEED
pip install iterabledata[geophysical]

# Microsoft Access (.mdb/.accdb)
pip install iterabledata[access]

# R fst frames (requires a suitable fst/rfst binding)
pip install iterabledata[fst]

# OpenTelemetry JSON and Protobuf export profiles
pip install iterabledata[otlp]

# Lakehouse table formats (Delta, Iceberg, Lance, Hudi, DuckLake)
pip install iterabledata[lakehouse]

# Apache Paimon (tables + Row + Mosaic file formats; separate from lakehouse)
pip install iterabledata[paimon]
# Or individually:
# pip install iterabledata[paimon-table]
# pip install iterabledata[paimon-row]
# pip install iterabledata[paimon-mosaic]
# pip install iterabledata[ducklake]

# Individual format extras (one per format family), for example:
pip install iterabledata[avro]     # Apache Avro
pip install iterabledata[npy]      # NumPy .npy/.npz
pip install iterabledata[ods]      # OpenDocument spreadsheets
pip install iterabledata[rdata]    # R RData/RDS
pip install iterabledata[ics]      # iCalendar
# Also available: ubj, vcf, capnp, thrift, fbs, edn, hocon, der, bencode, ldif, hdf5, xml, rdf

# All optional dependencies
pip install iterabledata[all]

AI 功能（[ai]）：支持使用 OpenAI、OpenRouter、Ollama、LMStudio 或 Perplexity 生成 AI 文档。

数据库引擎（[db]）：支持以可迭代数据源的形式只读访问数据库。支持 PostgreSQL、ClickHouse、MySQL/MariaDB、Microsoft SQL Server、SQLite、MongoDB 和 Elasticsearch/OpenSearch。包含便捷组合：

• [db-sql]：仅 SQL 数据库（PostgreSQL、ClickHouse、MySQL、MSSQL）
• [db-nosql]：仅 NoSQL 数据库（MongoDB、Elasticsearch）

基因组格式（[bio]）：支持基因组 VCF/BCF、CRAM、BED、GFF3 和 GTF。比对格式使用 pysam，可能需要参考文件。

地理空间／科学类额外项：[geospatial] 支持 FileGDB 和 MapInfo MIF（以及现有的 GeoPackage/Shapefile 技术栈）。[lidar]、[mat] 和 [geophysical] 分别支持 LAS、MATLAB MAT 和 SEG-Y/GRIB2/miniSEED。许多结构化格式（XYZ、CIF、PDB、ASCII Grid、CZML、EDI、WebDataset、Lotus WK1）无需额外依赖。

湖仓（[lakehouse]）：Delta Lake、Apache Iceberg、Lance、Apache Hudi 和 DuckLake。Delta、Iceberg 和 DuckLake 支持有界写入；目前 Hudi 为只读。

Paimon（[paimon]）：Apache Paimon 仓库表以及 Row 和 Mosaic 文件格式。如果只需要其中一种，可单独安装 [paimon-table]、[paimon-row] 或 [paimon-mosaic]。也可以仅通过 [ducklake] 使用 DuckLake。 有关这些功能的详细信息，请参阅 API 文档。

有关 AI Agent 和大模型工具，请参阅 llms.txt（简短索引）、llms-full.txt（可复制粘贴的配方）、可移植技能 skills/iterabledata/SKILL.md，以及 CONTRIBUTING.md。

AI 快速入门

通过 LM Studio 或 Ollama 使用本地大模型（无需 API 密钥）生成数据集文档，也可以使用 OpenAI：

from iterable.ai import doc

# Local (LM Studio on http://localhost:1234/v1)
documentation = doc.generate(
    "data.csv",
    provider="lmstudio",
    base_url="http://localhost:1234/v1",
    format="markdown",
)

# Or analyze structure + docs in one call
from iterable.ops import inspect

analysis = inspect.analyze("data.csv", autodoc=True, autodoc_provider="openai")
print(analysis["documentation"])

需要结构化、机器可读的输出？使用 generate_blocks() 获取独立的文档块（general、schema、quality、examples、statistics、agent_skill；另有可选的 codebook），以及汇总后的 Markdown：

from iterable.ai import doc

result = doc.generate_blocks(
    "data.csv",
    provider="openai",
    context={"title": "Sales 2025", "description": "Monthly sales export"},
    progress=lambda event: print(event.stage, event.detail),
)
print(result["blocks"]["schema"]["data"])     # structured JSON
print(result["blocks"]["agent_skill"]["markdown"])  # portable agent skill (YAML + Markdown)
print(result["full_document_markdown"])        # assembled markdown

agent_skill 块会生成可移植的技能文档（YAML frontmatter + Markdown），AI Agent 可加载此文档，以获取针对特定数据集的加载、查询和安全指导。

安装 AI 支持：pip install iterabledata[ai]。请参阅 examples/ai/ 和 docs/docs/api/ai.md。

嵌套架构和统计信息（可选）

from iterable.ops import schema, stats

sch = schema.infer("nested.jsonl", flatten_nested=True)
print(sch["fields"]["capital_city.lat"]["type"])

summary = stats.compute("nested.jsonl", flatten_nested=True)
print(summary["capital_city.lat"]["mean"])

对于多表工作簿／数据库，iterable.ai.table_profile.profile_selected_table() 会在行数／时间预算内分析指定的工作表／表（启用嵌套扁平化）。

格式目录（Agent）

from iterable.catalog import describe_format

info = describe_format("xml")
print(info["example_args"])  # {'tagname': 'item'}

完整导出：dev/formats.json 或 export_catalog(format="json")。请参阅 docs/docs/api/catalog.md。

快速入门

基本读取

from iterable import open_iterable

with open_iterable("data.csv.gz") as source:
    for row in source:
        print(row)

写入数据

from iterable import open_iterable

with open_iterable("output.jsonl.zst", mode="w") as dest:
    for item in my_data:
        dest.write(item)

使用示例

读取压缩 CSV 文件

from iterable import open_iterable

with open_iterable("data.csv.xz") as source:
    n = 0
    for row in source:
        n += 1
        if n % 1000 == 0:
            print(f"Processed {n} rows")

读取不同格式

from iterable import open_iterable

with open_iterable("data.jsonl") as source:
    for row in source:
        print(row)

with open_iterable("data.parquet") as source:
    for row in source:
        print(row)

with open_iterable("data.xml", iterableargs={"tagname": "item"}) as source:
    for row in source:
        print(row)

with open_iterable("data.xlsx") as source:
    for row in source:
        print(row)

# Read GeoJSON Text Sequence (streaming, one feature per line)
with open_iterable('features.geojsonl') as source:
    for feature in source:
        print(feature['properties'], feature['geometry'])

# Stream records from a TAR archive (members detected by filename)
with open_iterable('dataset.tar.gz', iterableargs={'members': '*.csv'}) as source:
    for row in source:
        print(row['_member'], row)

# Read genomic VCF (requires pip install iterabledata[bio])
with open_iterable('variants.vcf') as source:
    for variant in source:
        print(variant['chrom'], variant['pos'], variant['ref'], variant['alt'])

# Read genomic intervals (BED, GFF3, or GTF)
with open_iterable('genes.gff3') as source:
    for feature in source:
        print(feature['seqid'], feature['type'], feature['start'], feature['end'])

# Read a Zarr array (requires pip install iterabledata[zarr])
with open_iterable('signals.zarr', iterableargs={'array': 'values'}) as source:
    for row in source:
        print(row['value'])

# Read GeoParquet or FlatGeobuf (requires parquet/geospatial extras)
with open_iterable('roads.geoparquet') as source:
    for feature in source:
        print(feature.get('geometry'), feature.get('properties'))

# Read an OTLP JSON export (requires pip install iterabledata[otlp])
with open_iterable('telemetry.otlp.json') as source:
    for item in source:
        print(item['signal'], item['record'])

从数据库读取

from iterable import open_iterable

# Read from PostgreSQL database
with open_iterable(
    'postgresql://user:password@localhost:5432/mydb',
    engine='postgres',
    iterableargs={'query': 'users'}
) as source:
    for row in source:
        print(row)

# Read specific columns with filtering
with open_iterable(
    'postgresql://localhost/mydb',
    engine='postgres',
    iterableargs={
        'query': 'users',
        'columns': ['id', 'name', 'email'],
        'filter': 'active = TRUE'
    }
) as source:
    for row in source:
        print(row)

# Read from ClickHouse database
with open_iterable(
    'clickhouse://user:password@localhost:9000/analytics',
    engine='clickhouse',
    iterableargs={'query': 'events', 'settings': {'max_threads': 4}}
) as source:
    for row in source:
        print(row)

# Convert database to file
from iterable.convert import convert
convert(
    fromfile='postgresql://localhost/mydb',
    tofile='users.parquet',
    iterableargs={'engine': 'postgres', 'query': 'users'}
)

# Convert ClickHouse to Parquet
convert(
    fromfile='clickhouse://localhost:9000/analytics',
    tofile='events.parquet',
    iterableargs={'engine': 'clickhouse', 'query': 'events'}
)

格式检测和编码

from iterable import open_iterable
from iterable.helpers.detect import detect_file_type, detect_file_type_from_content
from iterable.helpers.utils import detect_encoding, detect_delimiter

# Detect file type and compression (uses filename extension)
result = detect_file_type('data.csv.gz')
print(f"Type: {result['datatype']}, Codec: {result['codec']}")

# Content-based detection (for files without extensions or streams)
with open('data.unknown', 'rb') as f:
    detection_result = detect_file_type_from_content(f)
    if detection_result:
        format_id, confidence, method = detection_result
        print(f"Detected format: {format_id} (confidence: {confidence:.2f}, method: {method})")

# open_iterable() automatically uses content-based detection as fallback
# Works with files without extensions, streams, or incorrect extensions
with open_iterable('data.unknown') as source:  # Detects from content
    for row in source:
        print(row)

# Detect encoding for CSV files
encoding_info = detect_encoding('data.csv')
print(f"Encoding: {encoding_info['encoding']}, Confidence: {encoding_info['confidence']}")

# Detect delimiter for CSV files
delimiter = detect_delimiter('data.csv', encoding=encoding_info['encoding'])

# Open with detected settings
source = open_iterable('data.csv', iterableargs={
    'encoding': encoding_info['encoding'],
    'delimiter': delimiter
})

错误处理

IterableData 提供完善的异常层次结构和可配置的错误处理：

from iterable import open_iterable
from iterable.exceptions import (
    FormatDetectionError,
    FormatNotSupportedError,
    FormatParseError,
    ReadError,
    CodecError,
    IterableDataError,
)

# Basic exception handling
try:
    with open_iterable('data.unknown') as source:
        for row in source:
            process(row)
except FormatDetectionError as e:
    print(f"Could not detect format: {e.reason}")
    # Try with explicit format or check file content
except FormatNotSupportedError as e:
    print(f"Format '{e.format_id}' not supported: {e.reason}")
    # Install missing dependencies or use different format
except FormatParseError as e:
    print(f"Failed to parse {e.format_id} format")
    if e.position:
        print(f"Error at position: {e.position}")
except ReadError as e:
    print(f"Read failed: {e}")
except IterableDataError as e:
    print(f"Library error: {e}")
except CodecError as e:
    print(f"Compression error with {e.codec_name}: {e.message}")
    # Check file integrity or try different codec
except Exception as e:
    print(f"Unexpected error: {e}")

可配置的错误策略：控制如何处理格式错误的记录：

# Skip malformed records and continue processing
with open_iterable(
    'data.csv',
    iterableargs={'on_error': 'skip', 'error_log': 'errors.log'}
) as src:
    for row in src:
        process(row)  # Only processes valid rows

# Warn on errors but continue processing
with open_iterable(
    'data.jsonl',
    iterableargs={'on_error': 'warn', 'error_log': 'errors.log'}
) as src:
    for row in src:
        process(row)  # Warnings logged, processing continues

# Default: raise exceptions immediately (existing behavior)
with open_iterable('data.csv', iterableargs={'on_error': 'raise'}) as src:
    for row in src:
        process(row)

不会静默读取空内容：在默认策略（on_error='raise'）下，格式错误的非空文件会引发 FormatParseError，而不是返回零条记录。使用 on_error='skip' 或 'warn' 可显式容忍错误记录。

Pickle 安全性：取消 pickle 序列化会执行任意代码。读取 pickle 文件时会发出警告，除非传入 trust=True：

with open_iterable('data.pickle', iterableargs={'trust': True}) as source:
    for row in source:
        process(row)

错误日志记录：包含上下文信息（文件名、行号、字节偏移量、错误消息、原始行）的结构化 JSON 日志。

完整的异常参考请参阅异常层次结构文档。

查询格式功能

from iterable.helpers.capabilities import (
    get_format_capabilities,
    get_capability,
    list_all_capabilities
)

# Get all capabilities for a format
caps = get_format_capabilities("csv")
print(f"CSV readable: {caps['readable']}")
print(f"CSV writable: {caps['writable']}")
print(f"CSV supports totals: {caps['totals']}")
print(f"CSV supports tables: {caps['tables']}")

# Query a specific capability
is_writable = get_capability("json", "writable")
has_totals = get_capability("parquet", "totals")
supports_tables = get_capability("xlsx", "tables")

# List capabilities for all formats
all_caps = list_all_capabilities()
for format_id, capabilities in all_caps.items():
    if capabilities.get("tables"):
        print(f"{format_id} supports multiple tables")

格式转换

from iterable import open_iterable
from iterable.convert import convert

# Simple format conversion
convert('input.jsonl.gz', 'output.parquet')

# Convert with options
convert(
    'input.csv.xz',
    'output.jsonl.zst',
    iterableargs={'delimiter': ';', 'encoding': 'utf-8'},
    batch_size=10000
)

# Convert and flatten nested structures
convert(
    'input.jsonl',
    'output.csv',
    is_flatten=True,
    batch_size=50000
)

通过原子写入确保生产安全

使用原子写入，确保输出文件不会处于部分写入状态：

from iterable.convert import convert
from iterable.pipeline import pipeline

# Convert with atomic writes (production-safe)
result = convert('input.csv', 'output.parquet', atomic=True)
# Output file only appears when conversion completes successfully

# Atomic writes in pipelines
pipeline(
    source=source,
    destination=destination,
    process_func=transform_func,
    atomic=True  # Ensures destination file is only created on success
)

优点：防止因崩溃、中断或进程中途失败而导致的数据损坏。发生故障时会保留原始文件。

批量文件转换

使用 glob 模式、目录或文件列表一次转换多个文件：

from iterable.convert import bulk_convert

# Convert all CSV files matching glob pattern
result = bulk_convert('data/raw/*.csv.gz', 'data/processed/', to_ext='parquet')

# Convert with custom filename pattern
result = bulk_convert('data/*.csv', 'output/', pattern='{name}.parquet')

# Convert entire directory
result = bulk_convert('data/raw/', 'data/processed/', to_ext='parquet')

# Access results
print(f"Converted {result.successful_files}/{result.total_files} files")
print(f"Total rows: {result.total_rows_out}")
print(f"Throughput: {result.throughput:.0f} rows/second")

# Check individual file results
for file_result in result.file_results:
    if file_result.success:
        print(f"✓ {file_result.source_file}: {file_result.result.rows_out} rows")
    else:
        print(f"✗ {file_result.source_file}: {file_result.error}")

功能：容错（单个文件失败时继续处理）、聚合指标、使用占位符（{name}、{stem}、{ext}）灵活命名输出文件。

进度跟踪与指标

通过回调、进度条和结构化指标跟踪转换和管道的进度：

from iterable.convert import convert
from iterable.pipeline import pipeline

# Progress callback for conversions
def progress_cb(stats):
    print(f"Progress: {stats['rows_read']} rows read, "
          f"{stats['rows_written']} rows written, "
          f"{stats.get('elapsed', 0):.2f}s elapsed")

# Convert with progress tracking
result = convert(
    'input.csv',
    'output.parquet',
    progress=progress_cb,
    show_progress=True  # Also shows tqdm progress bar
)

# Access conversion metrics
print(f"Converted {result.rows_out} rows in {result.elapsed_seconds:.2f}s")
print(f"Read {result.bytes_read} bytes, wrote {result.bytes_written} bytes")

# Pipeline with progress and metrics
result = pipeline(
    source=source,
    destination=destination,
    process_func=transform_func,
    progress=progress_cb  # Progress callback
)

# Access pipeline metrics (supports both attribute and dict access)
print(f"Processed {result.rows_processed} rows")
print(f"Throughput: {result.throughput:.0f} rows/second")
print(f"Exceptions: {result.exceptions}")
# Backward compatible: result['rec_count'] also works

功能：实时进度回调、使用 tqdm 自动显示进度条、结构化指标对象（ConversionResult、PipelineResult）。

使用管道处理数据

from iterable import open_iterable
from iterable.pipeline import pipeline

source = open_iterable('input.parquet')
destination = open_iterable('output.jsonl.xz', mode='w')

def transform_record(record, state):
    """Transform each record"""
    # Add processing logic
    out = {}
    for key in ['name', 'email', 'age']:
        if key in record:
            out[key] = record[key]
    return out

def progress_callback(stats, state):
    """Called every trigger_on records"""
    print(f"Processed {stats['rec_count']} records, "
          f"Duration: {stats.get('duration', 0):.2f}s")

def final_callback(stats, state):
    """Called when processing completes"""
    print(f"Total records: {stats['rec_count']}")
    print(f"Total time: {stats['duration']:.2f}s")

result = pipeline(
    source=source,
    destination=destination,
    process_func=transform_record,
    trigger_func=progress_callback,
    trigger_on=1000,
    final_func=final_callback,
    start_state={},
    atomic=True  # Use atomic writes for production safety
)

# Access pipeline metrics
print(f"Throughput: {result.throughput:.0f} rows/second")

source.close()
destination.close()

手动使用格式和编解码器

from iterable.datatypes.jsonl import JSONLinesIterable
from iterable.datatypes.bsonf import BSONIterable
from iterable.codecs.gzipcodec import GZIPCodec
from iterable.codecs.lzmacodec import LZMACodec

# Read gzipped JSONL
read_codec = GZIPCodec('input.jsonl.gz', mode='r', open_it=True)
reader = JSONLinesIterable(codec=read_codec)

# Write LZMA compressed BSON
write_codec = LZMACodec('output.bson.xz', mode='wb', open_it=False)
writer = BSONIterable(codec=write_codec, mode='w')

for row in reader:
    writer.write(row)

reader.close()
writer.close()

云存储支持

直接从云对象存储（S3、GCS、Azure）读取和写入数据：

from iterable import open_iterable

# Read from S3
with open_iterable('s3://my-bucket/data/events.csv') as source:
    for row in source:
        print(row)

# Read compressed file from GCS
with open_iterable('gs://my-bucket/data/events.jsonl.gz') as source:
    for row in source:
        process(row)

# Write to Azure Blob Storage
with open_iterable(
    'az://my-container/output/results.jsonl',
    mode='w',
    iterableargs={'storage_options': {'connection_string': '...'}}
) as dest:
    dest.write({'name': 'Alice', 'age': 30})
    dest.write({'name': 'Bob', 'age': 25})

支持的提供商：

• Amazon S3：s3:// 和 s3a:// 方案
• Google Cloud Storage：gs:// 和 gcs:// 方案
• Azure Blob Storage：az://、abfs:// 和 abfss:// 方案

安装：pip install iterabledata[cloud]

注意：DuckDB 引擎不支持云存储 URI；请使用 engine='internal'（默认值）。

使用 DuckDB 引擎和下推优化

DuckDB 引擎可通过高级优化实现高性能查询：

from iterable import open_iterable

# Basic DuckDB usage
source = open_iterable('data.csv.gz', engine='duckdb')
total = source.totals()  # Fast counting
for row in source:
    print(row)
source.close()

# Column projection pushdown (only read specified columns)
with open_iterable(
    'data.csv',
    engine='duckdb',
    iterableargs={'columns': ['name', 'age']}  # Reduces I/O and memory
) as src:
    for row in src:
        process(row)

# Filter pushdown (filter at database level)
with open_iterable(
    'data.csv',
    engine='duckdb',
    iterableargs={'filter': "age > 18 AND status = 'active'"}
) as src:
    for row in src:
        process(row)

# Combined column projection and filtering
with open_iterable(
    'data.parquet',
    engine='duckdb',
    iterableargs={
        'columns': ['name', 'age', 'email'],
        'filter': 'age > 18'
    }
) as src:
    for row in src:
        process(row)

# Direct SQL query support
with open_iterable(
    'data.parquet',
    engine='duckdb',
    iterableargs={
        'query': 'SELECT name, age FROM read_parquet(\'data.parquet\') WHERE age > 18 ORDER BY age DESC LIMIT 100'
    }
) as src:
    for row in src:
        process(row)

支持的格式：CSV、JSONL、NDJSON、JSON、Parquet
支持的编解码器：GZIP、ZStandard（.zst）
优点：通过数据库级优化减少 I/O、降低内存使用量并加快处理速度

批量操作

from iterable import open_iterable

source = open_iterable('input.jsonl')
destination = open_iterable('output.parquet', mode='w')

# Read and write in batches for better performance
batch = []
for row in source:
    batch.append(row)
    if len(batch) >= 10000:
        destination.write_bulk(batch)
        batch = []

# Write remaining records
if batch:
    destination.write_bulk(batch)

source.close()
destination.close()

面向性能的转换

当两端均为 Parquet 或 Arrow/Feather 时，请使用原生批次。如果支持，选择操作会下推至列式读取器；否则，转换会安全地回退到常规行/批量循环：

from iterable.convert import BatchSelection, convert

convert(
    'events.parquet',
    'events-copy.parquet',
    use_native_batch=True,
    selection=BatchSelection(columns=('event_id', 'created_at'), batch_size=8192),
    toiterableargs={'row_group_size': 32768},
)

对于一般压缩工作负载，请使用均衡配置。CPU 密集型管道请选择 fast，归档输出请选择 max：

with open_iterable('events.jsonl.zst', mode='w', codecargs={'profile': 'fast'}) as dest:
    dest.write_bulk(records)

有关选择、内存、行组和回退行为，请参阅原生批次、编解码器配置 和性能指南。

使用 Excel 文件

from iterable import open_iterable
from iterable.ai.fileinfo import open_table

# Read Excel file (specify sheet or page)
xls_file = open_iterable('data.xlsx', iterableargs={'page': 0})

for row in xls_file:
    print(row)
xls_file.close()

# Open a named sheet (uses page index under the hood)
sheet = open_table('data.xlsx', 'Sheet2')
for row in sheet:
    print(row)
sheet.close()

XML 处理

from iterable import open_iterable

# Parse XML with specific tag name
xml_file = open_iterable(
    'data.xml',
    iterableargs={
        'tagname': 'book',
        'prefix_strip': True  # Strip XML namespace prefixes
    }
)

for item in xml_file:
    print(item)
xml_file.close()

DataFrame 桥接

将可迭代数据转换为 Pandas、Polars 或 Dask DataFrame：

from iterable import open_iterable

# Convert to Pandas DataFrame
with open_iterable('data.csv.gz') as source:
    df = source.to_pandas()
    print(df.head())

# Chunked processing for large files
with open_iterable('large_data.csv') as source:
    for df_chunk in source.to_pandas(chunksize=100_000):
        # Process each chunk
        result = df_chunk.groupby('category').sum()
        process_chunk(result)

# Convert to Polars DataFrame
with open_iterable('data.csv.gz') as source:
    df = source.to_polars()
    print(df.head())

# Convert to Dask DataFrame (single file)
with open_iterable('data.csv.gz') as source:
    ddf = source.to_dask()
    result = ddf.groupby('category').sum().compute()

# Multi-file Dask DataFrame (automatic format detection)
from iterable.helpers.bridges import to_dask

ddf = to_dask(['file1.csv', 'file2.jsonl', 'file3.parquet'])
result = ddf.groupby('category').sum().compute()

注意：DataFrame 桥接需要可选依赖。请使用以下命令安装：

pip install iterabledata[dataframes]  # All DataFrame libraries
# Or individually:
pip install pandas
pip install polars
pip install "dask[dataframe]"

类型提示与类型安全

IterableData 为现代 Python 开发提供完整的类型注解和带类型的辅助函数：

from iterable import open_iterable
from iterable.helpers.typed import as_dataclasses, as_pydantic
from dataclasses import dataclass
from pydantic import BaseModel

# Type aliases for better code readability
from iterable.types import Row, IterableArgs, CodecArgs

# Convert to dataclasses for type safety
@dataclass
class Person:
    name: str
    age: int
    email: str | None = None

with open_iterable('people.csv') as source:
    for person in as_dataclasses(source, Person):
        # Full IDE autocomplete and type checking
        print(person.name, person.age)

# Convert to Pydantic models with validation
class PersonModel(BaseModel):
    name: str
    age: int
    email: str | None = None

with open_iterable('people.jsonl') as source:
    for person in as_pydantic(source, PersonModel, validate=True):
        # Automatic schema validation
        print(person.name, person.age)
        # Access as Pydantic model with all features

优势：

• 公共 API 均提供完整的类型注解
• py.typed 标记文件支持 mypy、pyright 和其他类型检查器
• 带类型的辅助函数可提供 IDE 自动补全和类型安全
• Pydantic 验证可及早发现架构问题

安装：运行 pip install iterabledata[pydantic] 以支持 Pydantic

进阶：将压缩 XML 转换为 Parquet

from iterable.datatypes.xml import XMLIterable
from iterable.datatypes.parquet import ParquetIterable
from iterable.codecs.bz2codec import BZIP2Codec

# Read compressed XML
read_codec = BZIP2Codec('data.xml.bz2', mode='r')
reader = XMLIterable(codec=read_codec, tagname='page')

# Write to Parquet with schema adaptation
writer = ParquetIterable(
    'output.parquet',
    mode='w',
    use_pandas=False,
    adapt_schema=True,
    batch_size=10000
)

batch = []
for row in reader:
    batch.append(row)
    if len(batch) >= 10000:
        writer.write_bulk(batch)
        batch = []

if batch:
    writer.write_bulk(batch)

reader.close()
writer.close()

API 参考

主要函数

open_iterable(filename, mode='r', engine='internal', codecargs={}, iterableargs={})

打开文件并返回一个可迭代对象。

参数：

• filename（str）：文件路径（支持本地文件和云存储 URI：s3://、gs://、az://）
• mode（str）：文件模式（'r' 表示读取，'w' 表示写入）
• engine（str）：处理引擎（'internal' 或 'duckdb'）
• codecargs（dict）：编解码器初始化参数
• iterableargs（dict）：可迭代对象初始化参数
  • columns（list[str]）：对于 DuckDB 引擎，仅读取指定列（下推优化）
  • filter（str | callable）：对于 DuckDB 引擎，在数据库级别筛选行（SQL 字符串或 Python 可调用对象）
  • query（str）：对于 DuckDB 引擎，执行自定义 SQL 查询（只读）
  • on_error（str）：错误处理策略（'raise'、'skip' 或 'warn'）
  • error_log（str | file-like）：用于结构化错误日志的路径或文件对象
  • storage_options（dict）：云存储身份验证选项

**返回值：**与检测到的文件类型对应的可迭代对象

detect_file_type(filename)

根据文件名检测文件类型和压缩编解码器。

**返回值：**包含 success、datatype 和 codec 键的字典

convert(fromfile, tofile, iterableargs={}, toiterableargs={}, scan_limit=1000, batch_size=50000, silent=True, is_flatten=False, use_totals=False, progress=None, show_progress=False, atomic=False, use_native_batch=False, selection=None, strict_native=False) -> ConversionResult

在不同格式之间转换数据。

参数：

• fromfile（str）：源文件路径
• tofile（str）：目标文件路径
• iterableargs（dict）：读取源文件的选项
• toiterableargs（dict）：写入目标文件的选项
• scan_limit（int）：用于检测架构的扫描记录数
• batch_size（int）：批量操作的批次大小
• silent（bool）：不输出进度信息
• is_flatten（bool）：展平嵌套结构
• use_totals（bool）：使用总数跟踪进度（如果可用）
• progress（callable）：可选的回调函数，接收进度统计字典
• show_progress（bool）：使用 tqdm 显示进度条（如果可用）
• atomic（bool）：写入临时文件，并在成功后以原子方式重命名
• use_native_batch（bool）：请求在两端均支持时进行原生列式批次传输
• selection（BatchSelection 或 dict）：可选的列、行范围、切片、表或后端谓词选择
• strict_native（bool）：原生批处理或请求的选择不受支持时抛出异常，而不是回退

返回值：ConversionResult 对象，包含：

• rows_in（int）：读取的总行数
• rows_out（int）：写入的总行数
• elapsed_seconds（float）：转换耗时
• bytes_read（int | None）：读取的字节数（如果可用）
• bytes_written（int | None）：写入的字节数（如果可用）
• errors（list[Exception]）：遇到的错误列表

bulk_convert(source, destination, pattern=None, to_ext=None, **kwargs) -> BulkConversionResult

使用 glob 模式、目录或文件列表一次转换多个文件。

参数：

• source (str)：glob 模式、目录路径或文件路径
• destination (str)：输出目录或文件名模式
• pattern (str)：带有占位符（{name}、{stem}、{ext}）的文件名模式
• to_ext (str)：替换文件扩展名（例如，'parquet'）
• **kwargs：convert() 函数的所有参数

**返回值：**包含以下内容的 BulkConversionResult 对象：

• total_files (int)：已处理的文件总数
• successful_files (int)：成功转换的文件数
• failed_files (int)：转换失败的文件数
• total_rows_in (int)：所有文件读取的总行数
• total_rows_out (int)：所有文件写入的总行数
• total_elapsed_seconds (float)：总转换时间
• file_results (list[FileConversionResult])：每个文件的处理结果
• errors (list[Exception])：遇到的所有错误
• throughput (float | None)：每秒行数

pipeline(source, destination, process_func, trigger_func=None, trigger_on=1000, final_func=None, reset_iterables=True, skip_nulls=True, start_state=None, debug=False, batch_size=1000, progress=None, atomic=False) -> PipelineResult

执行数据处理工作流。

参数：

• source (BaseIterable)：要读取的源可迭代对象
• destination (BaseIterable | None)：要写入的目标可迭代对象
• process_func (callable)：处理每条记录的函数
• trigger_func (callable | None)：处理期间定期调用的函数
• trigger_on (int)：每隔多少条记录调用一次触发函数
• final_func (callable | None)：处理完成后调用的函数
• reset_iterables (bool)：处理前重置可迭代对象
• skip_nulls (bool)：跳过 process_func 返回的 None 结果
• start_state (dict | None)：初始状态字典
• debug (bool)：抛出异常，而不是捕获异常
• batch_size (int)：写入前批处理的记录数
• progress (callable | None)：用于更新进度的可选回调函数
• atomic (bool)：如果目标是文件，则使用原子写入

**返回值：**包含以下内容的 PipelineResult 对象：

• rows_processed (int)：已处理的总行数
• elapsed_seconds (float)：处理时间
• throughput (float | None)：每秒行数
• exceptions (int)：遇到的异常数
• nulls (int)：空值结果数
• 为向后兼容，支持通过属性（result.rows_processed）和字典访问方式（result['rec_count']）进行访问

可迭代对象方法

所有可迭代对象都支持：

• read(skip_empty=True) -> Row - 读取单条记录
• read_bulk(num=DEFAULT_BULK_NUMBER) -> list[Row] - 读取多条记录
• write(record) - 写入单条记录
• write_bulk(records) - 写入多条记录
• reset() - 将迭代器重置到开头
• close() - 关闭文件句柄
• to_pandas(chunksize=None) - 转换为 pandas DataFrame（可选分块处理）
• to_polars(chunksize=None) - 转换为 Polars DataFrame（可选分块处理）
• to_dask(chunksize=1000000) - 转换为 Dask DataFrame
• list_tables(filename=None) -> list[str] | None - 列出可用的表/工作表/数据集
• has_tables() -> bool - 检查格式是否支持多个表

辅助函数

as_dataclasses(iterable, dataclass_type, skip_empty=True) -> Iterator[T]

将可迭代对象中的基于字典的行转换为数据类实例。

参数：

• iterable (BaseIterable)：要从中读取行的可迭代对象
• dataclass_type (type[T])：要将行转换为的数据类类型
• skip_empty (bool)：是否跳过空行

**返回值：**数据类实例的迭代器

as_pydantic(iterable, model_type, skip_empty=True, validate=True) -> Iterator[T]

将可迭代对象中的字典行转换为 Pydantic 模型实例。

参数：

• iterable (BaseIterable)：读取行的可迭代对象
• model_type (type[T])：要将行转换为的 Pydantic 模型类型
• skip_empty (bool)：是否跳过空行
• validate (bool)：是否根据模型架构验证行

返回： Pydantic 模型实例的迭代器

引发： 如果未安装 pydantic，则引发 ImportError

to_dask(files, chunksize=1000000, **iterableargs) -> DaskDataFrame

将多个文件转换为统一的 Dask DataFrame，并自动检测格式。

参数：

• files (str | list[str])：单个文件路径或文件路径列表
• chunksize (int)：每个分区的行数
• **iterableargs：为每个文件传递给 open_iterable() 的其他参数

返回： 包含所有文件数据的 Dask DataFrame

引发： 如果未安装 dask 或 pandas，则引发 ImportError

引擎

内部引擎（默认）

内部引擎使用纯 Python 实现支持所有格式，并支持所有文件类型和压缩编码。

DuckDB 引擎

DuckDB 引擎可为支持的格式提供高性能查询能力：

• 格式：CSV、JSONL、NDJSON、JSON
• 编码：GZIP、ZStandard (.zst)
• 功能：快速查询、总数计数、类 SQL 操作

打开文件时使用 engine='duckdb'：

source = open_iterable('data.csv.gz', engine='duckdb')

示例目录

更多完整示例请参阅 examples 目录：

• simplewiki/ - 处理 Wikipedia XML 转储

更多示例和测试

全面的用法示例和测试用例请参阅 tests 目录。

贡献者：使用 pytest --verbose 运行完整测试套件。性能回归门禁默认关闭，需选择启用：

pytest tests/test_performance_regression.py -m performance --no-cov

在发生合理的性能变更后，有意重新生成基线：

pytest tests/test_performance_regression.py -m performance --no-cov --update-baselines

开发约定请参阅 AGENTS.md。

AI 集成指南

IterableData 可与 AI 平台和框架集成，实现智能数据处理：

• AI 框架 - 与 LangChain、CrewAI 和 AutoGen 集成
  • 创建用于读取数据和转换格式的工具
  • 架构推断和数据质量分析
  • 用于数据处理的多 Agent 工作流
• OpenAI - 直接集成 OpenAI API（GPT-4、GPT-3.5 等）
  • 函数调用和 Assistants API
  • 结构化输出，确保结果一致
  • 自然语言数据分析和转换
• Claude - Anthropic Claude AI 集成
  • 支持工具的 Claude API 集成
  • 智能数据分析和架构推断
  • 在 AI 指导下转换格式
  • 数据质量评估和文档编写
• Gemini - Google Gemini AI 集成
  • 自然语言数据分析
  • 在 AI 指导下智能转换格式
  • 架构文档编写和数据质量评估
  • 函数调用集成

这些指南提供了模式、示例和最佳实践，帮助将 IterableData 的统一数据接口与 AI 能力相结合。

相关项目

此库是 Dateno 开源项目的一部分，也用于以下项目：

• undatum - 命令行数据处理工具
• datacrafter - 数据处理 ETL 引擎

许可证

IterableData 是 Dateno 开源项目的一部分。

• 代码采用 MIT License 许可。版权所有 © 2022–2026 Dateno。
• 文档和数据（包括此 README、文档站点和测试夹具）采用 CC BY 4.0 许可。

请参阅许可证页面，了解署名指南。

贡献

欢迎贡献！请随时提交拉取请求或提出问题。

更新日志

详细版本历史请参阅 CHANGELOG.md。

Version 1.0.23 (2026-08-18)

• MCP 注册表：缩短 server.json 描述，使其符合注册表的 100 个字符限制

Version 1.0.22 (2026-08-18)

• 编码 Agent：规范化 open_iterable / convert 导入、llms-full.txt、可移植技能和操作指南
• 发现机制：MCP server.json、/.well-known/llms.txt、提示词评估；在带标签的版本发布时通过 OIDC 发布到 MCP Registry
• CI：修复 Windows 文件锁定和 CSV 换行问题；基础安装会干净地跳过可选额外组件

Version 1.0.21 (2026-08-10)

• 嵌套架构/统计信息：在 schema.infer() / stats.compute() 中选择启用 flatten_nested=True，以使用 capital_city.lat 这样的点分路径
• 有界表分析：为多表数据源提供 iterable.ai.table_profile.profile_selected_table()，支持行数/时间预算
• Excel/SQLite 加固：正确打开命名工作表、跳过空标题行、恢复异常的 XLSX 维度；SQLite 优先以只读方式打开，并为表名加引号

Version 1.0.20 (2026-08-07)

• Agent 技能块（agent_skill）：默认的 generate_blocks() 现在会包含一份可移植的 Agent 技能文档（YAML 前置内容 + Markdown），其中包含数据集事实、工作流和安全指南
• 更安全的用法示例：示例 / 旧版 autodoc 提示词要求使用 python/r/sql，针对 dataset 表执行 SQL，并遵循只读限制
• 架构示例：将嵌套提供方的 example 值强制转换为字符串，以便进行结构化输出验证

Version 1.0.18 (2026-07-22)

• 开放数据格式（实验性）：FileGDB、MapInfo MIF、ASCII Grid、E00、LAS、BAG、CZML、XYZ、CIF、PDB、MATLAB MAT、SEG-Y、GRIB2、miniSEED、EDI、Access MDB、Lotus 1-2-3、WebDataset、R fst、HDT、IATI
• 额外组件：[lidar]、[mat]、[geophysical]、[access]、[fst]（以及现有的 [geospatial]、[hdf5]、[xml]、[rdf]）
• 新增湖仓格式：Apache Paimon Row/Mosaic 文件和 Paimon 目录表；DuckLake（ducklake / lakehouse 额外组件）
• 湖仓写入：支持 Delta Lake 和 Iceberg 的有界创建/追加/覆盖；支持 DuckLake 和 Paimon 表写入；暂不支持 Hudi 写入
• 额外组件（湖仓）：[paimon]、[paimon-table]、[paimon-row]、[paimon-mosaic] 和 [ducklake]；DuckLake 已并入 [lakehouse]
• 文档：扩充所有格式的占位页面；更新格式索引和侧边栏；修正 Delta/Iceberg 写入文档

Version 1.0.17 (2026-07-16)

• 性能：有界 Parquet/Arrow/Lance 批处理、共享行/批量游标、缓存转换总量，以及可选启用的原生列式批量转换。
• 压缩：提供 fast、balanced 和 max 编解码器配置，并显示有效设置诊断信息。
• 格式支持：Zarr、GeoParquet、FlatGeobuf、CRAM、BED、GFF3/GTF，以及 OTLP JSON/Protobuf 配置。
• 仓库质量：版本化能力元数据、发行包内容检查、夹具隔离、可选格式系列的 CI 覆盖，以及 OIDC 发布指南。

版本 1.0.16（2026-07-15）

• 新增格式：GeoJSON 文本序列（geojsonseq）、TAR 容器（tar）、基因组 VCF/BCF（genomic_vcf，bio extra）
• 安全性：安全防范 XXE 的 XML 解析、采用 AST 白名单的筛选表达式、对 pickle trust=True 的确认机制
• 流式处理：Snappy/LZO 流式编解码器；适用于 Shapefile、Arrow、Lance、Delta、Iceberg 的惰性／批量读取器
• 错误策略：默认情况下，格式错误的输入会引发类型化错误；open_iterable() 会向上抛出 IterableDataError 子类
• 质量：在 CI 中加入性能回归检查门禁；重构核心入口；增强测试的弹性

版本 1.0.15（2026-07-04）

• 基础安装即可导入：import iterable 不再需要可选的 BSON 或 Pydantic 依赖