IterableData

Iterable Data

Iterable Data は、一貫したイテレーター形式のインターフェースを使って、データファイルを行単位で読み書きするための Python ライブラリです。csv.DictReader に似た統一 API を提供し、CSV、JSON、Parquet、XML など、さらに多くの形式に対応しています。

このライブラリを使うと、複雑なネストされたデータ構造を保持したまま（フラット化が必要な 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 アーカイブからレコードをストリーミング
• 型ヒントと型安全性：dataclass と Pydantic モデル向けの型付きヘルパー関数を含む、完全な型注釈
• レイクハウステーブル：Delta Lake、Iceberg、DuckLake の読み書き、Hudi の読み取り、Apache Paimon テーブルの実験的サポート、Row／Mosaic ファイル形式

サポート対象のファイル形式

コア形式

• JSON - 標準的な JSON ファイル
• JSONL/NDJSON - JSON Lines 形式（1 行につき 1 つの JSON オブジェクト）
• JSON-LD - データ連携のための JSON（RDF 形式）
• CSV/TSV - カンマ区切りおよびタブ区切りの値
• Annotated 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 - ジオメトリと CRS の保持に対応した、メタデータ認識型の GeoParquet プロファイル
• Lance - ML とベクトル検索向けに最適化された最新の列指向形式
• Vortex - 高速なランダムアクセスに対応した最新の列指向形式
• Paimon Row - 行番号による O(1) アクセスに対応した Apache Paimon 行形式
• 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 - 科学データ向け Network Common Data Form
• CDF - NASA Common Data Format（宇宙科学）
• HDF5 - 階層データ形式
• Zarr - チャンク化配列ストア（zarr extra。試験的）
• XYZ - 分子／点座標テーブル
• CIF - 結晶構造情報ファイル（atom_site ループ。試験的）
• PDB - Protein Data Bank の 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 - Geography Markup Language
• KML - Keyhole Markup Language
• KMZ - KML の ZIP 圧縮形式（KML を含む ZIP アーカイブ）
• GPX - GPS Exchange Format（ウェイポイント、ルート、トラック）
• Shapefile - ESRI Shapefile 形式
• File Geodatabase - Fiona 経由の ESRI FileGDB レイヤー（geospatial extra。試験的）
• MapInfo MIF - MapInfo Interchange Format（geospatial extra。試験的）
• Esri ASCII Grid - セルまたは行レコードとしてのラスターグリッド（.asc）
• ArcInfo E00 - 交換用エクスポート（一部のみ試験的に対応）
• LAS - LiDAR 点群（lidar extra。試験的）
• BAG - Bathymetric Attributed Grid（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 - ヘッダー・辞書・トリプルをコンパクトにした RDF（rdf extra。実験的）

フィード形式

• Atom - Atom 配信形式
• RSS - リッチサイトサマリーのフィード形式

ネットワーク形式

• PCAP - パケットキャプチャ形式
• PCAPNG - PCAP 次世代形式

ログとイベント形式

• 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 - 人間向けに最適化された設定オブジェクト表記法
• 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 形式

グラフ形式

• GraphML - Graph Markup Language
• GEXF - Graph Exchange 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 - ML サンプル辞書ごとにグループ化された TAR シャード（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 Data Interchange Format
• 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 のファイル形式に対応します。必要な機能が 1 つだけの場合は、[paimon-table]、[paimon-row]、または [paimon-mosaic] を個別にインストールしてください。DuckLake 単独の [ducklake] も利用できます。 これらの機能の詳細については、API ドキュメントをご覧ください。

AI エージェントと LLM ツールについては、llms.txt（短いインデックス）、llms-full.txt（コピー＆ペースト可能なレシピ）、ポータブルスキル skills/iterabledata/SKILL.md、および CONTRIBUTING.md をご覧ください。

AI クイックスタート

LM Studio または Ollama を使って ローカル LLM でデータセットのドキュメントを生成できます（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 ブロックは、AI エージェントがデータセット固有の読み込み、クエリ、安全性に関するガイダンスを取得するための、ポータブルなスキルドキュメント（YAML のフロントマター＋Markdown）を出力します。

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() を使うと、行数／時間の予算内で名前を指定したシート／テーブルのプロファイルを作成できます（ネスト構造のフラット化が有効です）。

形式カタログ（エージェント向け）

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'）では、不正な空でないファイルを読み込むと、レコードを 0 件返すのではなく FormatParseError が発生します。不正なレコードを明示的に許容するには、on_error='skip' または 'warn' を使用してください。

Pickle の安全性：Pickle のデシリアライズでは任意のコードが実行されます。trust=True を渡さない限り、pickle ファイルの読み込み時に警告が表示されます。

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
)

メリット：クラッシュ、中断、処理途中の失敗によるデータ破損を防ぎます。失敗時も元のファイルは保持されます。

複数ファイルの一括変換

グロブパターン、ディレクトリ、またはファイルリストを使って、複数のファイルを一度に変換します。

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}")

機能：エラー耐性（1 つのファイルで失敗しても処理を続行）、集計されたメトリクス、プレースホルダー（{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）。

データ処理での Pipeline の使用

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 による検証で、スキーマの問題を早期に検出できます

インストール: Pydantic のサポートには pip install iterabledata[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 or 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)：1 秒あたりの行数

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)：1 秒あたりの行数
• exceptions (int)：発生した例外の数
• nulls (int)：null の結果の数
• 下位互換性のため、属性アクセス（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 ディレクトリ

より詳しい例については、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 との統合
  • データ読み込みと形式変換のためのツール作成
  • スキーマ推論とデータ品質分析
  • データ処理のためのマルチエージェントワークフロー
• 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の下でライセンスされています。Copyright © 2022–2026 Dateno.
• ドキュメントとデータ（この README、ドキュメントサイト、テスト用フィクスチャを含む）は、CC BY 4.0の下でライセンスされています。

クレジット表記のガイダンスについては、ライセンスページをご覧ください。

コントリビューション

コントリビューションを歓迎します！プルリクエストの送信や、イシューの作成をぜひお願いします。

変更履歴

詳細なバージョン履歴については、CHANGELOG.mdをご覧ください。

Version 1.0.23 (2026-08-18)

• MCP Registry：server.json の説明を、レジストリの上限である100文字に短縮

Version 1.0.22 (2026-08-18)

• コーディングエージェント：標準の 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_skill）：デフォルトの generate_blocks() に、データセットの事実、ワークフロー、安全性に関するガイダンスを含む、移植可能なエージェントスキルドキュメント（YAML frontmatter + 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 依存関係が不要になりました