Pure python MapFile, SetFile, ArrayFile

Author: matteobertozzi

python-hadoop/ArrayFileTest.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python
+
+from io.IntWritable import LongWritable, IntWritable
+from io import ArrayFile
+
+if __name__ == '__main__':
+    writer = ArrayFile.Writer('array-test', IntWritable)
+    writer.INDEX_INTERVAL = 16
+    for i in xrange(0, 100):
+        writer.append(IntWritable(1 + i * 10))
+    writer.close()
+
+    key = LongWritable()
+    value = IntWritable()
+    reader = ArrayFile.Reader('array-test')
+    while reader.next(key, value):
+        print key, value
+
+    print 'GET 8'
+    print reader.get(8, value)
+    print value
+    print
+
+    print 'GET 110'
+    print reader.get(110, value)
+    print
+
+    print 'GET 25'
+    print reader.get(25, value)
+    print value
+    print
+
+    print 'GET 55'
+    print reader.get(55, value)
+    print value
+    print
+
+    reader.close()

python-hadoop/MapFileTest.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python
+
+from io.IntWritable import LongWritable
+from io import MapFile
+
+if __name__ == '__main__':
+    writer = MapFile.Writer('map-test', LongWritable, LongWritable)
+    writer.INDEX_INTERVAL = 2
+    for i in xrange(0, 100, 2):
+        writer.append(LongWritable(i), LongWritable(i * 10))
+    writer.close()
+
+    key = LongWritable()
+    value = LongWritable()
+    reader = MapFile.Reader('map-test')
+    while reader.next(key, value):
+        print key, value
+
+    print 'GET CLOSEST'
+    key.set(8)
+    print reader.get(key, value)
+    print value
+    print
+
+    print 'GET 111'
+    key.set(111)
+    print reader.get(key, value)
+    print
+
+    key.set(25)
+    print 'SEEK 25 before'
+    print reader.getClosest(key, value, before=True)
+    print value
+    print
+
+    key.set(55)
+    print 'SEEK 55'
+    print reader.getClosest(key, value)
+    print value
+    print
+
+    reader.close()

python-hadoop/SetFileTest.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python
+
+from io.IntWritable import IntWritable
+from io import SetFile
+
+if __name__ == '__main__':
+    writer = SetFile.Writer('set-test', IntWritable)
+    writer.INDEX_INTERVAL = 16
+    for i in xrange(0, 100, 2):
+        writer.append(IntWritable(i * 10))
+    writer.close()
+
+    key = IntWritable()
+    reader = SetFile.Reader('set-test')
+    while reader.next(key):
+        print key
+
+    print 'GET 8'
+    key.set(8)
+    print reader.get(key)
+    print
+
+    print 'GET 120'
+    key.set(120)
+    print reader.get(key)
+    print
+
+    print 'GET 240'
+    key.set(240)
+    print reader.get(key)
+    print
+
+    print 'GET 550'
+    key.set(550)
+    print reader.get(key)
+    print
+
+    reader.close()

python-hadoop/io/ArrayFile.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python
+
+from IntWritable import LongWritable
+import MapFile
+
+class Writer(MapFile.Writer):
+    def __init__(self, path, value_class):
+        super(Writer, self).__init__(path, LongWritable, value_class)
+        self._count = 0
+
+    def append(self, value):
+        super(Writer, self).append(LongWritable(self._count), value)
+        self._count += 1
+
+class Reader(MapFile.Reader):
+    def __init__(self, path):
+        super(Reader, self).__init__(path)
+        self._key = LongWritable(0)
+
+    def seek(self, n):
+        if isinstance(n, LongWritable):
+            n = n.get()
+
+        self._key.set(n)
+        return super(Reader, self).seek(self._key)
+
+    def key(self):
+        return self._key.get()
+
+    def get(self, n, value):
+        self._key.set(n)
+        return(super(Reader, self).get(self._key, value))
+

python-hadoop/io/MapFile.py

@@ -1,7 +1,187 @@
 #!/usr/bin/env python
 
+import os
+
+from IntWritable import LongWritable
+import SequenceFile 
+
+INDEX_FILE_NAME = 'index'
+DATA_FILE_NAME = 'data'
+
 class Writer(object):
-    pass
+    INDEX_INTERVAL = 128
+
+    def __init__(self, dirname, key_class, value_class):
+        os.mkdir(dirname)
+
+        data_path = os.path.join(dirname, DATA_FILE_NAME)
+        self._data = SequenceFile.createWriter(data_path, key_class, value_class)
+
+        index_path = os.path.join(dirname, INDEX_FILE_NAME)
+        self._index = SequenceFile.createBlockWriter(index_path, key_class, LongWritable)
+
+        self._size = 0
+        self._last_index_pos = -1
+        self._last_index_nkeys = -4294967295
+
+    def close(self):
+        self._data.close()
+        self._index.close()
+
+    def append(self, key, value):
+        self._checkKey(key)
+
+        pos = self._data.getLength()
+        if self._size >= self._last_index_nkeys + self.INDEX_INTERVAL and pos > self._last_index_pos:
+            self._index.append(key, LongWritable(pos))
+            self._last_index_pos = pos
+            self._last_index_nkeys = self._size
+
+        self._data.append(key, value)
+        self._size += 1
+
+    def _checkKey(self, key):
+        pass
 
 class Reader(object):
-    pass
+    INDEX_SKIP = 0
+
+    def __init__(self, dirname):
+        self._data = SequenceFile.Reader(os.path.join(dirname, DATA_FILE_NAME))
+        self._index = SequenceFile.Reader(os.path.join(dirname, INDEX_FILE_NAME))
+        self._first_position = self._data.getPosition()
+        self._positions = []
+        self._keys = []
+
+    def close(self):
+        self._data.close()
+        self._index.close()
+
+    def getIndexInterval(self):
+        return self._index_interval
+
+    def setIndexInterval(self, interval):
+        self._index_interval = interval
+
+    def reset(self):
+        self._data.seek(self._first_position)
+
+    def midKey(self):
+        self._readIndex()
+        count = len(self._keys)
+        if count == 0:
+            return None
+        return keys[(count - 1) >> 1]
+
+    def finalKey(self, key):
+        original_position = self._data.getPosition()
+        try:
+            self._readIndex()
+            count = len(self._keys)
+            if count > 0:
+                self._data.seek(self._positions[count - 1])
+            else:
+                self._reset()
+            while self._data.nextKey(key):
+                continue
+        finally:
+            self._data.seek(original_position)
+
+    def seek(self, key):
+        return self._seekInternal(key) == 0
+
+    def next(self, key, value):
+        return self._data.next(key, value)
+
+    def get(self, key, value):
+        if self.seek(key):
+            self._data._getCurrentValue(value)
+            return value
+        return None
+
+    def getClosest(self, key, value, before=False):
+        c = self._seekInternal(key, before)
+        if (not before and c > 0) or (before and c < 0):
+            return None
+
+        self._data._getCurrentValue(value)
+        return self._next_key
+
+    def _readIndex(self):
+        if self._keys:
+            return
+
+        key_class = self._index.getKeyClass()
+
+        skip = self.INDEX_SKIP
+        position = LongWritable()
+        last_position = None
+        while True:
+            key = key_class()
+            if not self._index.next(key, position):
+                break
+
+            if skip > 0:
+                skip -= 1
+                continue
+
+            skip = self.INDEX_SKIP
+            if position.get() == last_position:
+                continue
+
+            self._positions.append(position.get())
+            self._keys.append(key)
+
+    def _seekInternal(self, key, before=None):
+        self._readIndex()
+
+        seek_index = self._indexSearch(key)
+        if seek_index < 0:
+            seek_index = -seek_index - 2
+
+        if seek_index == -1:
+            seek_position = self._first_position
+        else:
+            seek_position = self._positions[seek_index]
+
+        prev_position = -1
+        curr_position = seek_position
+
+        key_class = self._data.getKeyClass()
+        self._next_key = key_class()
+        
+        self._data.seek(seek_position)
+        while self._data.nextKey(self._next_key):
+            cmp = key.compareTo(self._next_key)
+            if cmp <= 0:
+                if before and cmp != 0:
+                    if prev_position == -1:
+                        self._data.seek(curr_position)
+                    else:
+                        self._data.seek(prev_position)
+                        self._data.nextKey(self._next_key)
+                        return 1
+                return cmp
+
+            if before:
+                prev_position = curr_position
+                curr_position = self._data.getPosition()
+
+        return 1
+
+    def _indexSearch(self, key):
+        high = len(self._keys) - 1
+        low = 0
+
+        while low <= high:
+            mid = (low + high) >> 1
+
+            cmp = self._keys[mid].compareTo(key)
+            if cmp < 0:
+                low = mid + 1
+            elif cmp > 0:
+                high = mid - 1
+            else:
+                return mid
+        return -(low + 1)
+

python-hadoop/io/SetFile.py

@@ -1,9 +1,12 @@
 #!/usr/bin/env python
 
-import NullWritable
+from NullWritable import NullWritable
 import MapFile
 
 class Writer(MapFile.Writer):
+    def __init__(self, path, key_class):
+        super(Writer, self).__init__(path, key_class, NullWritable)
+
     def append(self, key):
         return super(Writer, self).append(key, NullWritable())
 
@@ -13,7 +16,6 @@ def next(self, key):
 
     def get(self, key):
         if self.seek(key):
-            self.next(key)
-            return key
+            return self._next_key
         return None