#728 jsonl plugin Pull request opened by samhita-alla This PR adds jsonl (json lines) file and directory types, backed by

orjson

for fast serialization and optional

zstd

compression. pip install flyteplugins-jsonl # For Arrow RecordBatch support pip install 'flyteplugins-jsonl[arrow]'

JsonlFile

A single JSONL file. Inherits from

flyte.io.File

so it works with remote storage, upload/download and the Flyte type engine out of the box. from flyteplugins.jsonl import JsonlFile env = flyte.TaskEnvironment( image=flyte.Image().from_debian_base().with_pip_packages("flyteplugins-jsonl") ) # Async read @env.task async def process(f: JsonlFile): async for record in f.iter_records(): print(record) # Async write @env.task async def create() -> JsonlFile: f = JsonlFile.new_remote("data.jsonl") async with f.writer() as w: await w.write({"key": "value"}) return f # Sync write @env.task async def create_sync() -> JsonlFile: f = JsonlFile.new_remote("data.jsonl") with f.writer_sync() as w: w.write({"key": "value"}) return f

JsonlDir

A directory of sharded JSONL files (

part-00000.jsonl

,

part-00001.jsonl

, etc.). Inherits from

flyte.io.Dir

. Supports automatic shard rotation on write and transparent cross-shard iteration on read. from flyteplugins.jsonl import JsonlDir # Write with automatic sharding @env.task async def create() -> JsonlDir: d = JsonlDir.new_remote("output_shards") async with d.writer(max_records_per_shard=10_000) as w: for i in range(50_000): await w.write({"id": i}) return d # Read across all shards @env.task async def process(d: JsonlDir): async for record in d.iter_records(): print(record) flyteorg/flyte-sdk