Expects a coordinate WorkUnit for DragNet and runs the following
    steps:

    1. scans all dossiers at the *folder* level and assembles feature
    vectors for each folder -- see `make_feature`

    2. trains a multinomial naive Bayes classifier that treats each
    *folder* as a classifier target.

    3. sample the corpus by scanning up to `max_sample` and applying
    the classifier to each item to get an approx "size" of the Folder

    4. Bootstrap by treating those classifier predictions as truth
    data and extract the learned features that are predictive as new
    query strings.

    5. Put the data in kvlayer for webservice end point to return to
    polling client -- see dossier.models.routes

def worker(work_unit, max_sample=1000):
    '''Expects a coordinate WorkUnit for DragNet and runs the following
    steps:

    1. scans all dossiers at the *folder* level and assembles feature
    vectors for each folder -- see `make_feature`

    2. trains a multinomial naive Bayes classifier that treats each
    *folder* as a classifier target.

    3. sample the corpus by scanning up to `max_sample` and applying
    the classifier to each item to get an approx "size" of the Folder

    4. Bootstrap by treating those classifier predictions as truth
    data and extract the learned features that are predictive as new
    query strings.

    5. Put the data in kvlayer for webservice end point to return to
    polling client -- see dossier.models.routes

    '''
    if 'config' not in work_unit.spec:
        raise coordinate.exceptions.ProgrammerError(
            'could not run dragnet without global config')

    web_conf = Config()
    unitconf = work_unit.spec['config']
    with yakonfig.defaulted_config([coordinate, kvlayer, dblogger, web_conf],
                                   config=unitconf):

        labels = []
        D = list()

        label2fid = dict()

        rejects = set()
        keepers = set()

        # 1. make a classifier target for each *folder*, ignoring
        # subfolder structure
        FT = Folders(web_conf.kvlclient)
        for idx, fid in enumerate(FT.folders()):
            label2fid[idx] = fid
            for sid in FT.subfolders(fid):
                for cid, subtopic_id in FT.items(fid, sid):
                    fc = web_conf.store.get(cid)
                    if fc:
                        # NB: first call to make_feature
                        feat, _rejects, _keepers = make_feature(fc)
                    else:
                        _rejects = {}
                        _keepers = {}
                    D.append(feat)
                    labels.append(idx)
                    rejects.update(_rejects)
                    keepers.update(_keepers)
                    logger.info('fid=%r, observation: %r', fid, cid)

        # 2. Convert the StringCounters into an sklearn format and
        # train MultinomialNB
        logger.info('transforming...')
        v = DictVectorizer(sparse=False)
        X = v.fit_transform(D)
        logger.info('transform fit done.')

        labels = np.array(labels)

        # Fit the sklearn Bernoulli Naive Bayes classifer
        clf = MultinomialNB()
        clf.fit(X, labels)
        logger.info('fit MultinomialNB')

        # 3. Scan the corpus up to max_sample putting the items into
        # each target to get an approx "size" of the Folder
        counts = Counter()
        for cid, fc in islice(web_conf.store.scan(), max_sample):
            # build the same feature vector as the training process
            feat, _rejects, _keepers = make_feature(fc)
            X = v.transform([feat])
            # predict which folder it belongs in
            target = clf.predict(X[0])[0]
            # count the effective size of that folder in this sample
            counts[label2fid[target]] += 1

        logger.info('counts done')

        ## 4. Bootstrap by treating those classifier predictions as
        ## truth data and extract the learned features that are
        ## predictive as new query strings.
        clusters = []
        for idx in sorted(set(labels)):
            logger.debug('considering cluster: %d', idx)
            try:
                all_features = v.inverse_transform(clf.feature_log_prob_[idx])[0]
            except:
                logger.warn('beyond edge on cluster %d', idx)
                continue
            words = Counter(all_features)
            ordered = sorted(words.items(),
                             key=operator.itemgetter(1), reverse=True)
            filtered = []
            for it in ordered:
                if is_bad_token(it[0]): continue

                if is_username(it[0]):
                    logger.debug('%r is_username', it[0])
                #else:
                #    continue
                filtered.append(it)
                if len(filtered) > 100: # hard cutoff
                    break

            # normalize cluster size exponentially
            biggest = exp(filtered[0][1])
            # rescale all by biggest
            filtered = [(key, int(round(counts[label2fid[idx]] * exp(w) / biggest))) for key, w in filtered]
            # describe what we just figured out
            logger.info('%s --> %r', label2fid[idx], ['%s: %d' % it for it in filtered[:10]])

            # return build the JSON-serializable format for the
            # DragNet UI embedded inside SortingDesk
            cluster = []
            cluster.append({'caption': label2fid[idx],
                            'weight': counts[label2fid[idx]],
                            'folder_id': None,
                            })
            cluster += [{'caption': caption, 'weight': weight, 'folder_id': label2fid[idx]} for caption, weight in filtered if weight > 0]
            clusters.append(cluster)

        # 5. Put the data in kvlayer for webservice end point to
        # return to polling client
        web_conf.kvlclient.setup_namespace({'dragnet': (str,)})
        web_conf.kvlclient.put('dragnet', (('dragnet',), json.dumps({'clusters': clusters})))
        return dict(counts)