asyncstream

UNDER CONSTRUCTION

Simple library to compress/uncompress Async streams using file iterator and readers.

It supports the following compression format:

• gzip
• bzip2
• snappy
• zstandard
• parquet (experimental)
• orc (experimental)

Getting started

Install the library as follows:

pip install asyncstream

Compress a regular file to gzip (examples/simple_compress_gzip.py):

importasyncstreamimportasyncioasyncdefrun():
asyncwithasyncstream.open('samples/animals.txt', 'rb') asfd:
asyncwithasyncstream.open('samples/animals.txt.gz', compression='gzip') asgzfd:
asyncforlineinfd:
awaitgzfd.write(line)
if__name__=='__main__':
asyncio.run(run())

or you can also open from an async file descriptor using aiofiles (examples/simple_compress_gzip_with_aiofiles.py):

pip install aiofiles

And then run the following code:

importaiofilesimportasyncstreamimportasyncioasyncdefrun():
asyncwithaiofiles.open('examples/animals.txt', 'rb') asfd:
asyncwithaiofiles.open('/tmp/animals.txt.gz', 'wb') aswfd:
asyncwithasyncstream.open(wfd, 'wb', compression='gzip') asgzfd:
asyncforlineinfd:
awaitgzfd.write(line)
if__name__=='__main__':
asyncio.run(run())

You can also uncompress an S3 file on the fly using aiobotocore:

pip install aiobotocore

And then run the following code (examples/simple_uncompress_bzip2_from_s3.py):

importaiobotocoreimportasyncstreamimportasyncioasyncdefrun():
session=aiobotocore.get_session()
asyncwithsession.create_client('s3') ass3:
obj=awaits3.get_object(Bucket='test-bucket', Key='path/to/file.bz2')
asyncwithasyncstream.open(obj['Body'], 'rt', compression='bzip2') asfd:
asyncforlineinfd:
print(line)
if__name__=='__main__':
asyncio.run(run())

Convert a gzip file to a snappy file

importasyncstreamimportasyncioasyncdefrun():
asyncwithasyncstream.open('samples/animals.txt.gz', 'rb', compression='gzip') asinc_fd:
asyncwithasyncstream.open('samples/animals.txt.snappy', 'wb', compression='snappy') asoutc_fd:
asyncforlineininc_fd:
awaitoutc_fd.write(line)
if__name__=='__main__':
asyncio.run(run())

Use an async reader and writer to filter and update data on the fly

importasyncstreamimportasyncioasyncdefrun():
asyncwithasyncstream.open('/tmp/animals.txt.bz2', 'rb') asin_fd:
asyncwithasyncstream.open('/tmp/animals.txt.snappy', 'wb') asout_fd:
asyncwithasyncstream.reader(in_fd) asreader:
asyncwithasyncstream.writer(out_fd) aswriter:
asyncforname, color, ageinreader:
ifcolor!='white':
awaitwriter.writerow([name, color, age*2])
asyncio.run(run())

Simple parquet encoding

You will need to install pyarrow and pandas:

pip install pyarrow pandas

To compress using snappy, you can install snappy:

pip install snappy

The code below converts a csv file and convert it to parquet

importasyncstreamimportasyncioasyncdefrun():
asyncwithasyncstream.open('examples/animals.txt', 'rb') asfd:
asyncwithasyncstream.open('output.parquet', 'wb', encoding='parquet', compression='snappy') aswfd:
asyncwithasyncstream.writer(wfd) aswriter:
asyncforlineinfd:
awaitwriter.write(line)
asyncio.run(run())

Simple parquet decoding

importasyncstreamimportasyncioasyncdefrun():
asyncwithasyncstream.open('output.parquet', 'rb', encoding='parquet') asfd:
asyncwithasyncstream.reader(fd) asreader:
asyncforlineinreader:
print(line)
asyncio.run(run())

Simple orc encoding

importasyncstreamimportasyncioasyncdefrun():
asyncwithasyncstream.open('examples/animals.txt', 'rb') asfd:
asyncwithasyncstream.open('output.orc.snappy', 'wb', encoding='orc', compression='snappy') aswfd:
asyncwithasyncstream.writer(wfd) aswriter:
asyncforlineinfd:
awaitwriter.write(line)
asyncio.run(run())

Simple orc decoding

importasyncstreamimportasyncioasyncdefrun():
asyncwithasyncstream.open('output.orc.snappy', 'rb', encoding='orc') asfd:
asyncwithasyncstream.reader(fd) asreader:
asyncforlineinreader:
print(line)
asyncio.run(run())

Usage

asyncstream.open(afd: Union[str, AsyncBufferedIOBase], mode=None, encoding=None, compression=None, compress_level=-1)

Open an async file (using its filename or an AsyncBufferedIOBase) and compress or decompress on the fly depending on the mode (r or w).

Inputs:

• mode: r or w and t (text) or b (binary)
• encoding: None, parquet or orc
• compression: See the compression supported section
• compress_level: Optional if compression is used

The AsyncFileObj object returned has the following methods:

• flush(): used when open in write mode
• close(): close file descriptor and release resources. It is done automatically when the async with block is exited

asyncstream.reader(afd: AsyncFileObj, columns=Optional[Iterable[str]], column_types=Optional[Iterable[str]], has_header=False, sep=',', eol='\n')

Create an async reader using AsyncFileObj returned by the asyncstream.open method. It must be open in text mode (t).

Inputs:

• afd: AsyncFileObj created with asyncstream.open
• columns: optional list of column names to use. If it is not set and has_header is true, then it will use the first row as the column names
• column_types: optional list of column types (by default it will consider all the columns to be string)
• has_header: the file has the first line set as header
• sep: separator between values
• eol: end of line character

asyncstream.writer(afd: AsyncFileObj, columns: Optional[Iterable[str]] = None, column_types: Optional[Iterable[str]] = None, has_header: bool = True, sep=',', eol='\n')

Create an async writer using AsyncFileObj returned by the asyncstream.open method. It must be open in text mode (t).

Inputs:

• afd: AsyncFileObj created with asyncstream.open in text mode t and write w
• columns: optional list of column names to use. If it is not set and has_header is true, then it will use the first row as the column names
• column_types: optional list of column types (by default it will consider all the columns to be string)
• has_header: the file has the first line set as header
• sep: separator between values
• eol: end of line character

Compression supported

Compression	Status
gzip / zlib	✅
bzip2	✅
snappy	✅
zstd	✅

Parquet

Compression	Status
none	✅
brotli	✅
bzip2	❌
gzip	❌
snappy	✅
zstd	❌
zlib	✅

Orc

Compression	Status
none	✅
bzip2	❌
gzip / zlib	✅
snappy	✅
zlib	✅
zstd	✅