#Clean_redis this flow is here to remove user which are not matching pattern

@task
def ownBatcher(iterable, n):
    while True:
        batch = []
        for i in range(n):
            try:
                batch.append(next(iterable))
            except StopIteration:
                return batch
        if len(batch) != 0:
            yield batch


@task()
def filterList(userToFilter):
    pattern = '.*\.mydomain.net.*'
    users = []

    for userTuple in userToFilter:
        for user in userTuple:
            if user:
                if not re.match(pattern, user.decode("utf-8")):
                    users.append(user)
    return users

@task()
def removeUsers(redisCon, users):
    if(users):
        redisCon.delete(*users)

def main():

    with Flow("clean_redis") as flow:
       redisConnection = redis.Redis(host='127.0.0.1', port=6379, db=0)
       userBatch = ownBatcher(redisConnection.scan_iter('*'), 5000)
       usersToRemove = filterList.map(userBatch)
       removeUsers.map(unmapped(redisConnection), usersToRemove)

    flow.executor = LocalDaskExecutor()
    flow.run()

main()

redisConnection.scan_iter(match='*.mydomain.net.*'):