def bloomfilter(collection, on, column, capacity=3000, error_rate=0.01):
    """
    Filter collection on the `on` sequence by BloomFilter built by `column`

    :param collection:
    :param on: sequence or column name
    :param column: instance of Column
    :param capacity: numbers of capacity
    :type capacity: int
    :param error_rate: error rate
    :type error_rate: float
    :return: collection

    :Example:

    >>> df1 = DataFrame(pd.DataFrame({'a': ['name1', 'name2', 'name3', 'name1'], 'b': [1, 2, 3, 4]}))
    >>> df2 = DataFrame(pd.DataFrame({'a': ['name1']}))
    >>> df1.bloom_filter('a', df2.a)
           a  b
    0  name1  1
    1  name1  4
    """


    if not isinstance(column, Column):
        raise TypeError('bloomfilter can only filter on the column of a collection')

    # to make the class pickled right by the cloudpickle
    with open(os.path.join(path, 'lib', 'bloomfilter.py')) as bloomfilter_file:
        local = {}
        six.exec_(bloomfilter_file.read(), local)
        BloomFilter = local['BloomFilter']

        col_name = column.source_name or column.name

        on_name = on.name if isinstance(on, SequenceExpr) else on
        rand_name = '%s_%s'% (on_name, str(uuid.uuid4()).replace('-', '_'))
        on_col = collection._get_field(on).rename(rand_name)
        src_collection = collection
        collection = collection[collection, on_col]

        @output(src_collection.schema.names, src_collection.schema.types)
        class Filter(object):
            def __init__(self, resources):
                table = resources[0]

                bloom = BloomFilter(capacity, error_rate)
                for row in table:
                    bloom.add(str(getattr(row, col_name)))

                self.bloom = bloom

            def __call__(self, row):
                if str(getattr(row, rand_name)) not in self.bloom:
                    return
                return row[:-1]

        return collection.apply(Filter, axis=1, resources=[column.input, ])
def bloomfilter ( collection , on , column , capacity = 3000 , error_rate = 0.01 ) : if not isinstance ( column , Column ) : raise TypeError ( 'bloomfilter can only filter on the column of a collection' ) with open ( os . path . join ( path , 'lib' , 'bloomfilter.py' ) ) as bloomfilter_file : local = { } six . exec_ ( bloomfilter_file . read ( ) , local ) BloomFilter = local [ 'BloomFilter' ] col_name = column . source_name or column . name on_name = on . name if isinstance ( on , SequenceExpr ) else on rand_name = '%s_%s' % ( on_name , str ( uuid . uuid4 ( ) ) . replace ( '-' , '_' ) ) on_col = collection . _get_field ( on ) . rename ( rand_name ) src_collection = collection collection = collection [ collection , on_col ] @ output ( src_collection . schema . names , src_collection . schema . types ) class Filter ( object ) : def __init__ ( self , resources ) : table = resources [ 0 ] bloom = BloomFilter ( capacity , error_rate ) for row in table : bloom . add ( str ( getattr ( row , col_name ) ) ) self . bloom = bloom def __call__ ( self , row ) : if str ( getattr ( row , rand_name ) ) not in self . bloom : return return row [ : - 1 ] return collection . apply ( Filter , axis = 1 , resources = [ column . input , ] )
Filter collection on the `on` sequence by BloomFilter built by `column`

    :param collection:
    :param on: sequence or column name
    :param column: instance of Column
    :param capacity: numbers of capacity
    :type capacity: int
    :param error_rate: error rate
    :type error_rate: float
    :return: collection

    :Example:

    >>> df1 = DataFrame(pd.DataFrame({'a': ['name1', 'name2', 'name3', 'name1'], 'b': [1, 2, 3, 4]}))
    >>> df2 = DataFrame(pd.DataFrame({'a': ['name1']}))
    >>> df1.bloom_filter('a', df2.a)
           a  b
    0  name1  1
    1  name1  4