Help: snappy decompression broken?

[ambimorph]

I'm wondering if there is a version incompatibility.
Here's my traceback:

In [1]: from kafka import KafkaConsumer, codec

In [2]: codec.has_snappy()
Out[2]: True

In [3]: consumer = KafkaConsumer('xxx', bootstrap_servers =['xx.xxx.xx.xxx:xxxx']) # (substituted x's)

In [4]: x = consumer.next()
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-4-9399a37ddcd6> in <module>()
----> 1 x = consumer.next()

/opt/anaconda-2.3.0/lib/python2.7/site-packages/six.pyc in next(self)
    533
    534         def next(self):
--> 535             return type(self).__next__(self)
    536
    537     callable = callable

/opt/anaconda-2.3.0/lib/python2.7/site-packages/kafka/consumer/group.pyc in __next__(self)
    795         self._set_consumer_timeout()
    796         try:
--> 797             return next(self._iterator)
    798         except StopIteration:
    799             self._iterator = None

/opt/anaconda-2.3.0/lib/python2.7/site-packages/kafka/consumer/group.pyc in _message_generator(self)
    756                 continue
    757 
--> 758             for msg in self._fetcher:
    759                 yield msg
    760                 if time.time() > timeout_at:

/opt/anaconda-2.3.0/lib/python2.7/site-packages/six.pyc in next(self)
    533 
    534         def next(self):
--> 535             return type(self).__next__(self)
    536 
    537     callable = callable
/opt/anaconda-2.3.0/lib/python2.7/site-packages/kafka/consumer/group.pyc in __next__(self)
    795         self._set_consumer_timeout()
    796         try:
--> 797             return next(self._iterator)
    798         except StopIteration:
    799             self._iterator = None

/opt/anaconda-2.3.0/lib/python2.7/site-packages/kafka/consumer/group.pyc in _message_generator(self)
    756                 continue
    757 
--> 758             for msg in self._fetcher:
    759                 yield msg
    760                 if time.time() > timeout_at:

/opt/anaconda-2.3.0/lib/python2.7/site-packages/six.pyc in next(self)
    533 
    534         def next(self):
--> 535             return type(self).__next__(self)
    536 
    537     callable = callable

/opt/anaconda-2.3.0/lib/python2.7/site-packages/kafka/consumer/fetcher.pyc in __next__(self)
    437             self._iterator = self._message_generator()
    438         try:
--> 439             return next(self._iterator)
    440         except StopIteration:
    441             self._iterator = None

/opt/anaconda-2.3.0/lib/python2.7/site-packages/kafka/consumer/fetcher.pyc in _message_generator(self)
    405                 log.log(0, "Returning fetched records at offset %d for assigned"
    406                            " partition %s", position, tp)
--> 407                 for msg in self._unpack_message_set(tp, messages):
    408
    409                     # Because we are in a generator, it is possible for

/opt/anaconda-2.3.0/lib/python2.7/site-packages/kafka/consumer/fetcher.pyc in _unpack_message_set(self, tp, messages)
    354                 raise Errors.InvalidMessageError(msg)
    355             elif msg.is_compressed():
--> 356                 for record in self._unpack_message_set(tp, msg.decompress()):
    357                     yield record
    358             else:

/opt/anaconda-2.3.0/lib/python2.7/site-packages/kafka/protocol/message.pyc in decompress(self)
     70         elif codec == self.CODEC_SNAPPY:
     71             assert has_snappy(), 'Snappy decompression unsupported'
---> 72             raw_bytes = snappy_decode(self.value)
     73         elif codec == self.CODEC_LZ4:
     74             assert has_lz4(), 'LZ4 decompression unsupported'

/opt/anaconda-2.3.0/lib/python2.7/site-packages/kafka/codec.pyc in snappy_decode(payload)
    171             cursor += 4
    172             end = cursor + block_size
--> 173             out.write(snappy.decompress(byt[cursor:end]))
    174             cursor = end
    175

AttributeError: 'module' object has no attribute 'decompress'

[dpkp]

What snappy Python library and version do you have installed?

[ambimorph]

python-snappy 2.4.1

[dpkp]

can you try:

import snappy
print(snappy.__file__)
print(dir(snappy))

and post the output?

[ambimorph]

In [8]: print(snappy.__file__)
/opt/anaconda-2.3.0/lib/python2.7/site-packages/snappy/__init__.pyc

In [9]: print(dir(snappy))
['AbelianGroup', 'AlternatingKnotExteriors', 'CensusKnots', 'Crossing', 'CubicalNonorientableClosedCensus', 'CubicalNonorientableCuspedCensus', 'CubicalOrientableClosedCensus', 'CubicalOrientableCuspedCensus', 'CuspNeighborhood', 'CuspNeighborhoodHP', 'DTcodec', 'DirichletDomain', 'DirichletDomainHP', 'DodecahedralNonorientableClosedCensus', 'DodecahedralNonorientableCuspedCensus', 'DodecahedralOrientableClosedCensus', 'DodecahedralOrientableCuspedCensus', 'FXrays', 'FundamentalGroup', 'HTLinkExteriors', 'HolonomyGroup', 'HolonomyGroupHP', 'IcosahedralNonorientableClosedCensus', 'IcosahedralOrientableClosedCensus', 'IdentityBraid', 'InfinityTangle', 'Link', 'LinkExteriors', 'Manifold', 'ManifoldHP', 'NonalternatingKnotExteriors', 'NonorientableClosedCensus', 'NonorientableCuspedCensus', 'OctahedralNonorientableCuspedCensus', 'OctahedralOrientableCuspedCensus', 'OrientableClosedCensus', 'OrientableCuspedCensus', 'RationalTangle', 'SnapPeaFatalError', 'SnapPy', 'SnapPyHP', 'SnapPy_doc', 'Strand', 'SymmetryGroup', 'Tangle', 'TetrahedralNonorientableCuspedCensus', 'TetrahedralOrientableCuspedCensus', 'Triangulation', 'TriangulationHP', 'ZeroTangle', '_ManifoldHP', '_ManifoldLP', '_TriangulationHP', '_TriangulationLP', '__all__', '__builtins__', '__doc__', '__file__', '__name__', '__package__', '__path__', '__version__', '_link_exterior', '_link_lookup_DT', '_within_sage', 'all_translations', 'canonical_retriangulation', 'cusp_translations', 'database', 'database_objects', 'db_utilities', 'decorated_isosig', 'filedialog', 'isometry_signature', 'link_objects', 'manifolds', 'method', 'mfld_class', 'number', 'pari', 'ptolemy', 'random_link', 'release_info', 'sage_helper', 'snap', 'snappy', 'sys', 'twister', 'verify', 'version']

[dpkp]

Looks like you are getting interference from snap-py: https://snap.stanford.edu/snappy/

[ambimorph]

Ah! Thank you so much! I uninstalled them both, reinstalled python-snappy, and voila:

In [1]: import snappy

In [2]: print(snappy.__file__)
/opt/anaconda-2.3.0/lib/python2.7/site-packages/snappy.pyc

In [3]: print(dir(snappy))
['CompressError', 'CompressedLengthError', 'InvalidCompressedInputError', 'StreamCompressor', 'StreamDecompressor', 'UncompressError', '_CHUNK_MAX', '_COMPRESSED_CHUNK', '_COMPRESSI
ON_THRESHOLD', '_IDENTIFIER_CHUNK', '_RESERVED_SKIPPABLE', '_RESERVED_UNSKIPPABLE', '_STREAM_IDENTIFIER', '_STREAM_TO_STREAM_BLOCK_SIZE', '_UNCOMPRESSED_CHUNK', '__builtins__', '__d
oc__', '__file__', '__name__', '__package__', '_compress', '_crc32c', '_masked_crc32c', '_uncompress', 'cmdline_main', 'compress', 'decompress', 'isValidCompressed', 'stream_compres
s', 'stream_decompress', 'struct', 'uncompress']

In [4]: from kafka import KafkaConsumer

In [5]: consumer = KafkaConsumer('xxx', bootstrap_servers =['xx.xxx.xx.xxx:xxxx'])

In [6]: x = consumer.next()

In[7]:

[dpkp]

Fantastic!

[zinking]

thanks same issue for me.
on my mac, I did brew install snappy as well as pip install snappy
the lateral is not needed.

[rajeevanair]

Hello,
I am in the same boat. I have a .csv file which is snappy compressed. I am trying to uncompress it with snappy.uncompress() function. However, I am not able to find the uncompress function in the snappy package. Kindly, help.

Thanks,
Rajeev A. Nair

[bhaskar-c]

on debian:
sudo apt-get install libsnappy-dev
followed by
pip install python-snappy

solved the issue for me

[mahailuo]

on debian:
sudo apt-get install libsnappy-dev
followed by
pip install python-snappy

solved the issue for me

I have solved the problem with this method，Thanks!