It's quite kludgy (self referencing, etc; just put a few minutes writing it, nothing really elegant), but it does what you want if you're still interested in using gzip instead of zlib directly.
Basically, GzipWrap is a (very limited) file-like object that produces a gzipped file out of a given iterable (e.g., a file-like object, a list of strings, any generator...)
Of course, it produces binary so there was no sense in implementing "readline".
You should be able to expand it to cover other cases or to be used as an iterable object itself.
from gzip import GzipFile
class GzipWrap(object):
# input is a filelike object that feeds the input
def __init__(self, input, filename = None):
self.input = input
self.buffer = ''
self.zipper = GzipFile(filename, mode = 'wb', fileobj = self)
def read(self, size=-1):
if (size < 0) or len(self.buffer) < size:
for s in self.input:
self.zipper.write(s)
if size > 0 and len(self.buffer) >= size:
self.zipper.flush()
break
else:
self.zipper.close()
if size < 0:
ret = self.buffer
self.buffer = ''
else:
ret, self.buffer = self.buffer[:size], self.buffer[size:]
return ret
def flush(self):
pass
def write(self, data):
self.buffer += data
def close(self):
self.input.close()
Answer from Ricardo Cárdenes on Stack OverflowIt's quite kludgy (self referencing, etc; just put a few minutes writing it, nothing really elegant), but it does what you want if you're still interested in using gzip instead of zlib directly.
Basically, GzipWrap is a (very limited) file-like object that produces a gzipped file out of a given iterable (e.g., a file-like object, a list of strings, any generator...)
Of course, it produces binary so there was no sense in implementing "readline".
You should be able to expand it to cover other cases or to be used as an iterable object itself.
from gzip import GzipFile
class GzipWrap(object):
# input is a filelike object that feeds the input
def __init__(self, input, filename = None):
self.input = input
self.buffer = ''
self.zipper = GzipFile(filename, mode = 'wb', fileobj = self)
def read(self, size=-1):
if (size < 0) or len(self.buffer) < size:
for s in self.input:
self.zipper.write(s)
if size > 0 and len(self.buffer) >= size:
self.zipper.flush()
break
else:
self.zipper.close()
if size < 0:
ret = self.buffer
self.buffer = ''
else:
ret, self.buffer = self.buffer[:size], self.buffer[size:]
return ret
def flush(self):
pass
def write(self, data):
self.buffer += data
def close(self):
self.input.close()
Here is a cleaner, non-self-referencing version based on Ricardo Cárdenes' very helpful answer.
from gzip import GzipFile
from collections import deque
CHUNK = 16 * 1024
class Buffer (object):
def __init__ (self):
self.__buf = deque()
self.__size = 0
def __len__ (self):
return self.__size
def write (self, data):
self.__buf.append(data)
self.__size += len(data)
def read (self, size=-1):
if size < 0: size = self.__size
ret_list = []
while size > 0 and len(self.__buf):
s = self.__buf.popleft()
size -= len(s)
ret_list.append(s)
if size < 0:
ret_list[-1], remainder = ret_list[-1][:size], ret_list[-1][size:]
self.__buf.appendleft(remainder)
ret = ''.join(ret_list)
self.__size -= len(ret)
return ret
def flush (self):
pass
def close (self):
pass
class GzipCompressReadStream (object):
def __init__ (self, fileobj):
self.__input = fileobj
self.__buf = Buffer()
self.__gzip = GzipFile(None, mode='wb', fileobj=self.__buf)
def read (self, size=-1):
while size < 0 or len(self.__buf) < size:
s = self.__input.read(CHUNK)
if not s:
self.__gzip.close()
break
self.__gzip.write(s)
return self.__buf.read(size)
Advantages:
- Avoids repeated string concatenation, which would cause the entire string to be copied repeatedly.
- Reads a fixed CHUNK size from the input stream, instead of reading whole lines at a time (which can be arbitrarily long).
- Avoids circular references.
- Avoids misleading public "write" method of GzipCompressStream(), which is really only used internally.
- Takes advantage of name mangling for internal member variables.
» pip install gzip-stream
Using GCS, cloudstorage.open(filename, 'r') will give you a read-only file-like object (earlier created similarly but with 'w':-) which you can use, a chunk at a time, with the standard Python library's zlib module, specifically a zlib.decompressobj, if, of course, the GS object was originally created in the complementary way (with a zlib.compressobj).
Alternatively, for convenience, you can use the standard Python library's gzip module, e.g for the reading phase something like:
compressed_flo = cloudstorage.open('objname', 'r')
uncompressed_flo = gzip.GzipFile(fileobj=compressed_flo,mode='rb')
csvReader = csv.reader(uncompressed_flo)
and vice versa for the earlier writing phase, of course.
Note that when you run locally (with the dev_appserver), the GCS client library uses local disk files to simulate GCS -- in my experience that's good for development purposes, and I can use gsutil or other tools when I need to interact with "real" GCS storage from my local workstation... GCS is for when I need such interaction from my GAE app (and for developing said GAE app locally in the first place:-).
So, you have gzipped files stored on GCS. You can process the data stored on GCS in a stream-like fashion. That is, you can download, unzip, and process simultaneously. This avoids
- to have the unzipped file on disk
- to have to wait until the download is complete before being able to process the data.
gzip files have a small header and footer, and the body is a compressed stream, consisting of a series of blocks, and each block is decompressable on its own. Python's zlib package helps you with that!
Edit: This is example code for how to decompress and analzye a zlib or gzip stream chunk-wise, purely based on zlib:
import zlib
from collections import Counter
def stream(filename):
with open(filename, "rb") as f:
while True:
chunk = f.read(1024)
if not chunk:
break
yield chunk
def decompress(stream):
# Generate decompression object. Auto-detect and ignore
# gzip wrapper, if present.
z = zlib.decompressobj(32+15)
for chunk in stream:
r = z.decompress(chunk)
if r:
yield r
c = Counter()
s = stream("data.gz")
for chunk in decompress(s):
for byte in chunk:
c[byte] += 1
print c
I tested this code with an example file data.gz, created with GNU gzip.
Quotes from http://www.zlib.net/manual.html:
windowBits can also be greater than 15 for optional gzip decoding. Add 32 to windowBits to enable zlib and gzip decoding with automatic header detection, or add 16 to decode only the gzip format (the zlib format will return a Z_DATA_ERROR). If a gzip stream is being decoded, strm->adler is a crc32 instead of an adler32.
and
Any information contained in the gzip header is not retained [...]