Importing bitarray library into SparkContext

Question

I am trying to import the bitarray library into a SparkContext. https://pypi.python.org/pypi/bitarray/0.8.1.

To do this I have zipped up the contexts in the bit array folder and then tried to add it to my python files. However even after I push the library to the nodes my RDD cannot find the library. Here is my code

zip bitarray.zip bitarray-0.8.1/bitarray/*

// Check the contents of the zip file 

unzip -l bitarray.zip
Archive:  bitarray.zip
  Length      Date    Time    Name
---------  ---------- -----   ----
   143455  2015-11-06 02:07   bitarray/_bitarray.so
     4440  2015-11-06 02:06   bitarray/__init__.py
     6224  2015-11-06 02:07   bitarray/__init__.pyc
    68516  2015-11-06 02:06   bitarray/test_bitarray.py
    78976  2015-11-06 02:07   bitarray/test_bitarray.pyc
---------                     -------
   301611                     5 files

then in spark

import os 

# Environment
import findspark
findspark.init("/home/utils/spark-1.6.0/")

import pyspark
sparkConf = pyspark.SparkConf()

sparkConf.set("spark.executor.instances", "2") 
sparkConf.set("spark.executor.memory", "10g")
sparkConf.set("spark.executor.cores", "2")

sc = pyspark.SparkContext(conf = sparkConf)

from pyspark.sql import SQLContext
from pyspark.sql.types import *
from pyspark.sql import HiveContext
from pyspark.sql.types import StructType, StructField, StringType
from pyspark.sql.functions import udf

hiveContext = HiveContext(sc)

PYBLOOM_LIB = '/home/ryandevera/pybloom.zip'
sys.path.append(PYBLOOM_LIB)
sc.addPyFile(PYBLOOM_LIB)

from pybloom import BloomFilter
f = BloomFilter(capacity=1000, error_rate=0.001)
x = sc.parallelize([(1,("hello",4)),(2,("goodbye",5)),(3,("hey",6)),(4,("test",7))],2)


def bloom_filter_spark(iterator):
    for id,_ in iterator:
        f.add(id)
    yield (None, f)

x.mapPartitions(bloom_filter_spark).take(1)

This yields the error -

ImportError: pybloom requires bitarray >= 0.3.4

I am not sure where I am going wrong. Any help would be greatly appreciated!

Answer 1

Probably the simplest thing you is to create and distribute egg files. Assuming you've downloaded and unpacked source files from PyPI and set PYBLOOM_SOURCE_DIR and BITARRAY_SOURCE_DIR variables:

cd $PYBLOOM_SOURCE_DIR
python setup.py bdist_egg
cd $BITARRAY_SOURCE_DIR
python setup.py bdist_egg

In PySpark add:

from itertools import chain
import os
import glob

eggs = chain.from_iterable([
  glob.glob(os.path.join(os.environ[x], "dist/*")) for x in   
  ["PYBLOOM_SOURCE_DIR", "BITARRAY_SOURCE_DIR"]
])

for egg in eggs: sc.addPyFile(egg)

Problem is that BloomFilter object cannot be properly serialized so if you want to use it you'll have to either patch it or extract bitarrays and pass these around:

def buildFilter(iter):
    bf = BloomFilter(capacity=1000, error_rate=0.001)
    for x in iter:
        bf.add(x)
    return [bf.bitarray]

rdd = sc.parallelize(range(100))
rdd.mapPartitions(buildFilter).reduce(lambda x, y: x | y)