TsFileデータの読み込み

TsFileは、時系列データ向けに設計されたカラムナファイル形式であり、Apache IoTDBのネイティブなストレージ層として使用されています。Parquetのような汎用的なカラムナ形式と比較して、TsFileは時系列データモデル（タイムスタンプ、デバイス、メジャーメント）を認識し、ファイル全体をスキャンせずに時間範囲のプルーニングを可能にする内部の時間インデックスを維持します。

このローダーは、通常の1行1レコードのタブラー形式の慣例に従わないため、別のガイドとして提供されています。各出力行は1つのdeviceに対応し、測定ごとの値はArrowのlist<...>カラムとして返されます。マッピングの詳細は以下に説明します。

インストール

ローダーはtsfilePythonパッケージに依存しています:

pip install "tsfile>=2.3.0"

データモデルと出力レイアウト

ローダーはTsFileのテーブルモデルに従います。各テーブルの列は、次のいずれかです:

• TAG — 文字列型の識別子。device（つまり、単一の時系列ソース）は、TAG 値のタプルによって一意に識別されます。
• FIELD — 値が時間とともに変化する測定値。
• TIME — タイムスタンプ列。デフォルトではtimeと命名されます。

ローダーはデバイスごとに1行のデータセットを出力します。各行内では、timeカラムおよびすべてのFIELDカラムは、昇順にソートされたそのデバイスの完全な時系列を含むArrowのlist<...>カラムです。TAGカラムはスカラーのstringカラムとして表示されます。

具体的には、出力スキーマは以下の形式になります。

<tag_1>:    string
<tag_2>:    string                       # one column per TAG
...
time:       list<timestamp[unit, tz]>
<field_1>:  list<original_type>          # one column per FIELD
<field_2>:  list<original_type>
...

同じデバイスが分割の複数の入力ファイルに出現する場合、ファイルごとのチャンクは連結され、タイムスタンプ順にソートされた後、単一の行として出力されます。同じデバイスのタイムスタンプが重複すると、ValueErrorが発生します。

基本的な使用方法

単一のTsFileを読み込む:

>>> from datasets import load_dataset
>>> dataset = load_dataset("tsfile", data_files="my_data.tsfile")

ファイルをスプリットに明示的にマッピングする:

>>> dataset = load_dataset(
...     "tsfile",
...     data_files={"train": "train_data.tsfile", "test": "test_data.tsfile"},
... )

Hub上の例のデータセット

tsfile/lotsa_data にすぐに使える例が用意されています。.tsfile ファイルは自動的に認識されるため、data_files を指定せずにリポジトリ ID で読み込むことができます:

>>> from datasets import load_dataset
>>> dataset = load_dataset("tsfile/lotsa_data")
>>> dataset
DatasetDict({
    train: Dataset({
        features: ['timeseries_id', 'time', 'value'],
        num_rows: 91
    })
})

各行は1つのデバイスに対応します。timeseries_id列がデバイスを識別し、timeとvalueは、そのデバイスの完全な時系列データを含むlist<...>列です。

>>> row = dataset["train"][0]
>>> row["timeseries_id"]
'Bear_assembly_Angel'
>>> len(row["time"]), len(row["value"])
(8760, 8760)
>>> row["time"][:3]
[datetime.datetime(2017, 1, 1, 0, 0), datetime.datetime(2017, 1, 1, 1, 0), datetime.datetime(2017, 1, 1, 2, 0)]

テーブルの選択

TsFileには複数のテーブルを含めることができます。table_nameが省略された場合、最初の有効なファイル内で見つかった最初のテーブルが使用されます。検索は大文字と小文字を区別しません。

>>> dataset = load_dataset("tsfile", data_files="my_data.tsfile", table_name="sensor_data")

列の選択

columns は、読み込まれる FIELD 列を制限します。TAG 列と time 列は、デバイスとそのタイムラインを識別するために常に返されます。columns 内の TAG や time 列を指す名前は、黙って無視されます（通常どおり、1 回だけ出力されます）。すべてのファイルに存在しないフィールドに一致する名前は、すべて null のリスト列になります。

>>> dataset = load_dataset(
...     "tsfile",
...     data_files="my_data.tsfile",
...     columns=["temperature", "humidity"],
... )

時間範囲によるフィルタリング

start_time と end_time は両端を含む境界値です。どちらか一方を省略することもできます。境界値は TsFile の内部時間インデックスにプッシュダウンされるため、ディスクから読み込まれるのは該当するデータブロックのみです。両方の境界値には、以下のいずれかを指定できます。

• int — timestamp_unit 単位の生エポック値（デフォルトはミリ秒）;
• datetime.datetime — naiveな値はUTCとして解釈され、タイムゾーン付きの値はUTCに変換されます;
• datetime.date;
• ISO-8601 str、例: "2024-01-01T00:00:00";
• pyarrow.TimestampScalar.

>>> from datetime import datetime
>>> dataset = load_dataset(
...     "tsfile",
...     data_files="my_data.tsfile",
...     start_time=datetime(2023, 11, 14),
...     end_time="2023-11-15T00:00:00",
... )

ファイル間のスキーマの進化

異なるファイルが異なるカラムを公開している場合（例えば、後に新しいセンサーフィールドが導入された場合）、ローダーはすべてのFIELDカラムの和集合を取得し、欠損値をnullで埋めます。数値型のFIELDは、IoTDBのワイドニングルールに従って昇格されます（INT32 → INT64 → DOUBLE、INT32 → FLOAT → DOUBLE）。

>>> dataset = load_dataset("tsfile", data_files=["day1.tsfile", "day2.tsfile"])

読み取れないファイルの処理

デフォルトでは、読み取れない入力またはTsFile以外の入力がエラーを引き起こします。on_bad_filesを"warn"に設定するとログを記録して処理を続行し、"skip"に設定するとファイルを無言で破棄します。

>>> dataset = load_dataset("tsfile", data_files="data/*.tsfile", on_bad_files="skip")

タイムスタンプの単位とタイムゾーン

timestamp_unit (デフォルトは "ms"、IoTDB と一致) は、time 列の解像度と整数型時間境界の解釈を制御します。timestamp_tz は Arrow の timestamp 型にタイムゾーンを付与します。None (デフォルト) の場合、タイムゾーンなしの型が生成されます。

>>> dataset = load_dataset(
...     "tsfile",
...     data_files="my_data.tsfile",
...     timestamp_unit="us",
...     timestamp_tz="UTC",
... )

メモリとバッチ処理

メモリ使用量を制御するパラメータは2つあります:

• input_batch_size (デフォルト: 65_536) — TsFileReader.query_table から 1 つの Arrow バッチごとに取得する行の最大数です。単一のデバイスをストリーミング処理する際のピークメモリを制限します。
• output_batch_size (デフォルト: 32) — writer に渡される各 Arrow レコードバッチにパッキングされるデバイスの数です。値を小さくすると進捗レポートの応答性が向上し、大きくするとバッチごとのオーバーヘッドが軽減されます。

>>> dataset = load_dataset(
...     "tsfile",
...     data_files="large_data.tsfile",
...     input_batch_size=32_768,
...     output_batch_size=128,
... )

ピークメモリは、分割全体ではなく、分割全体にわたる単一デバイスのペイロードによって上限が設定されます。

パラメータの完全なリストについては、TsFileConfigを参照してください。