Litelink - 本地优先嵌入式流数据持久化库

项目简介

Litelink 是一个 Python 库，解决了数据捕获管道中最容易被手工实现出问题的环节：将数据流持久化写入 Parquet 文件，最终进入对象存储，无需守护进程、无需消息队列，也无需目录服务。append() 方法在数据行写入磁盘后立即返回，查询调用可在稍后任何时刻读取到这条数据。

Litelink 的核心理念是「嵌入式和本地优先」，设计目标是在单机环境（如树莓派）即可完整运行整个数据管道，同时支持与生产级对象存储（S3）无缝衔接，将数据归档到远程 Iceberg 表供其他分析引擎查询。

核心功能

1. 一次写入三重存储分层

Litelink 将数据写入路径分为三层：本地 SQLite 缓冲区（随每次提交持久化）、本地 Iceberg 表（滚动窗口，读请求直接命中本地文件）、远程 Iceberg 表（完整历史，异步上传到 S3）。三层各司其职，查询热路径完全不经过网络。

2. append() 同步返回

每行数据的 append() 调用在数据落盘后同步返回，无需等待后台进程或网络上传。这意味着写入延迟是可预测的，不受网络抖动或对象存储响应时间影响。

3. extend() 批量提交优化

extend() 方法将一组数据行在单次事务中提交，一次 fsync 完成整个批次而非逐行 fsync。该设计是写入吞吐量的核心杠杆，适合高频数据捕获场景。

4. 远程 Iceberg 归档

通过 litelink.snapshot() 接口可在任意支持 Iceberg 的查询引擎（如 DuckDB、Spark）中直接读取归档数据。归档表通过版本提示文件（version-hint.text）发布，查询引擎自行解析元数据，无需额外目录服务。

5. WAL 复制与故障恢复

litelink.restore() 可在另一台机器上重建日志，保留偏移窗口避免数据重放。通过 WAL 边车进程将写节点的 buffer.db 实时复制到备机，即使写节点宕机也不会丢失未归档的数据。

6. 无凭证本地化设计

凭证信息从不写入日志目录，库通过标准 AWS 凭证链（profile、实例元数据、SSO）从环境变量读取。这意味着在隔离环境中也可正常运行，且无需修改代码即可切换凭证来源。

7. DuckDB 集成开箱即用

litelink 内置 DuckDB 扩展，在 x86-64 和 ARM64 架构的 Linux 和 macOS 上预置了校验和验证的 litestream 二进制。运行 python -m litelink 即可检查目标机器环境是否满足运行时依赖。

8. Parquet 列式存储

所有写入数据以 Parquet 格式存储为列式压缩文件，文件大小可配置（target_seal_size），系统自动执行压缩、淘汰和过期清理（maintain）。

技术实现

Litelink 的架构设计源于一个实际生产案例的教训：某数据捕获系统积累了 125,884 个对象，其中 62.5% 小于 16 KiB，Parquet 文件平均仅 2 行，压缩任务从未被正确调度，内存缓冲区在 SIGKILL 时被清空。Litelink 的设计者从这个案例中提取了三条核心原则：

第一，每次提交必须真正落盘，不能依赖后台进程异步处理。第二，写入路径必须可中断、可恢复，网络故障不能导致数据丢失。第三，文件大小和压缩策略必须可配置和可调度，而非由写入频率被动决定。

数据流具体路径为：数据行 → 本地 SQLite 缓冲区（持久化 WAL）→ 本地 Iceberg Parquet 文件（达到 seal 大小时触发）→ 远程 S3 Iceberg 归档（异步同步）。每层都有独立的读写路径，层间通过 offset 机制对齐，防止数据重复或遗漏。

快速上手

第一步：安装。

pip install litelink
# 或
uv add litelink

第二步：定义 Schema。

import litelink
import pyarrow as pa

schema = pa.schema([
    pa.field("trade_id", pa.int64()),
    pa.field("event_ts", pa.int64()),
    pa.field("price", pa.float64()),
    pa.field("amount", pa.float64()),
])

第三步：创建日志并追加数据。

log = litelink.new("data", "trades", schema=schema, sort_by=("event_ts",))
log.append({"trade_id": 624438572, "event_ts": 1787772776240000,
            "price": 78501.62, "amount": 0.0076})

第四步：批量写入与维护。

log.extend(group_of_rows)
recent = log.scan(where="event_ts > 1787772776000000").read_all()
log.maintain()  # 压缩、淘汰、过期

第五步：跨机器读取归档数据。

import duckdb
con = duckdb.connect()
con.execute("CREATE SECRET (TYPE s3, PROVIDER credential_chain, REGION 'us-east-1');")
table = con.execute("""
  SELECT count(*), max(litelink_offset)
  FROM iceberg_scan('s3://bucket/prefix/trades',
  version_name_format = '%s%s.metadata.json')
""").arrow().read_all()

适用人群

Litelink 面向以下场景：金融数据捕获（交易数据、市场数据、物联网时序数据）需要本地持久化保证；数据工程团队希望减少 Kafka/Pulsar 等消息队列的运维复杂度；边缘计算场景需要在没有稳定网络的环境下完成数据捕获，回到中心后再同步归档；以及需要构建轻量级数据管道、避免引入重型基础设施的独立开发者和小型团队。

GitHub 地址：https://github.com/nhobin219/litelink