Загрузка данных TsFile

TsFile — это столбчатый формат файлов, разработанный для временных рядов и используемый в качестве нативного слоя хранения в Apache IoTDB. В отличие от универсальных столбчатых форматов, таких как Parquet, TsFile учитывает модель данных временных рядов (метки времени, устройства и измерения) и поддерживает внутренний временной индекс, который позволяет выполнять отсечение по диапазону времени без сканирования целых файлов.

Этот загрузчик вынесен в отдельное руководство, поскольку он не следует стандартному табличному формату «одна строка на запись»: каждая выходная строка соответствует одному устройству, а значения по каждому измерению возвращаются в виде столбцов Arrow list<...>. Соответствие описано подробно ниже.

Установка

Загрузчик зависит от пакета tsfile для Python:

pip install "tsfile>=2.3.0"

Модель данных и структура вывода

Загрузчик следует модели таблиц TsFile. Каждая колонка таблицы является одной из:

• TAG — идентификатор типа строки; кортеж значений TAG однозначно определяет устройство (т. е. источник одного временного ряда).
• FIELD — величина, значение которой изменяется во времени.
• TIME — столбец с меткой времени, по умолчанию называется time.

Загрузчик формирует по одной строке датасета на устройство. Внутри строки колонка 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:

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

Каждая строка соответствует одному устройству. Колонка TAG 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, молча игнорируются (они выводятся как обычно, один раз); имена, которые совпадают с полем, отсутствующим во всех файлах, становятся столбцами списков, заполненными значениями null.

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

Фильтрация по временному диапазону

start_time и end_time — это включительные границы; любая из них может быть опущена. Границы передаются во внутренний временной индекс TsFile, поэтому с диска читаются только соответствующие блоки данных. Обе границы принимают одно из значений:

• int — эпоха в единицах timestamp_unit (по умолчанию миллисекунды);
• datetime.datetime — значения без часового пояса интерпретируются как 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; None (по умолчанию) даёт тип без привязки к часовому поясу.

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

Память и пакетная обработка

Два параметра управляют использованием памяти:

• input_batch_size (по умолчанию 65_536) — максимальное количество строк, получаемых за один Arrow-батч из TsFileReader.query_table. Ограничивает пиковое потребление памяти при потоковой обработке на одном устройстве.
• output_batch_size (по умолчанию 32) — количество устройств, упаковываемых в каждый пакет записей Arrow, передаваемый в writer. Меньшие значения обеспечивают более оперативное отображение прогресса; большие значения снижают накладные расходы на каждый пакет.

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

Пиковое потребление памяти ограничено размером полезной нагрузки на одном устройстве в рамках разбиения, а не общим размером всего разбиения.

Список всех параметров см. в TsFileConfig.