## 词频统计 #

tokens = {}

for token in tokenize(text):
tokens[token] = tokens.get(token, 0) + 1

## 多进程版本 #

def parallel_apply(func,
iterable,
workers,
max_queue_size,
callback=None,
dummy=False):
"""多进程或多线程地将func应用到iterable的每个元素中。
注意这个apply是异步且无序的，也就是说依次输入a,b,c，但是
输出可能是func(c), func(a), func(b)。
参数：
dummy: False是多进程/线性，True则是多线程/线性；
callback: 处理单个输出的回调函数；
"""
if dummy:
from multiprocessing.dummy import Pool, Queue
else:
from multiprocessing import Pool, Queue
from six.moves import queue

in_queue, out_queue = Queue(max_queue_size), Queue()

def worker_step(in_queue, out_queue):
# 单步函数包装成循环执行
while True:
d = in_queue.get()
r = func(d)
out_queue.put(r)

# 启动多进程/线程
pool = Pool(workers, worker_step, (in_queue, out_queue))

if callback is None:
results = []

# 后处理函数
def process_out_queue():
out_count = 0
for _ in range(out_queue.qsize()):
d = out_queue.get()
out_count += 1
if callback is None:
results.append(d)
else:
callback(d)
return out_count

# 存入数据，取出结果
in_count, out_count = 0, 0
for d in iterable:
in_count += 1
while True:
try:
in_queue.put(d, block=False)
break
except queue.Full:
out_count += process_out_queue()
if in_count % max_queue_size == 0:
out_count += process_out_queue()

while out_count != in_count:
out_count += process_out_queue()

pool.terminate()

if callback is None:
return results


def _batch_texts():
texts = []
texts.append(text)
if len(texts) == 1000:
yield texts
texts = []
if texts:
yield texts

def _tokenize_and_count(texts):
tokens = {}
for text in texts:
for token in tokenize(text):
tokens[token] = tokens.get(token, 0) + 1

tokens = {}
def _total_count(result):
for k, v in result.items()
tokens[k] = tokens.get(k, 0) + v

# 10进程来完成词频统计
parallel_apply(
func=_tokenize_and_count,
iterable=_batch_texts(),
workers=10,
max_queue_size=200,
callback=_total_count,
)


## 原理分析 #

count = 0
tokens = {}
_tokens = {}

for token in tokenize(text):
_tokens[token] = _tokens.get(token, 0) + 1
count += 1
if count == 1000:
for k, v in _tokens.items():
tokens[k] = tokens.get(k, 0) + v
count = 0
_tokens = {}

for k, v in _tokens.items():
tokens[k] = tokens.get(k, 0) + v
{"ft_lang_label":"__label__zh","ft_lang_prob":0.623815,"math_prob":0.9634891,"size":4233,"snap":"2019-43-2019-47","text_gpt3_token_len":2418,"char_repetition_ratio":0.12627098,"word_repetition_ratio":0.11392405,"special_character_ratio":0.27096623,"punctuation_ratio":0.19379845,"nsfw_num_words":0,"has_unicode_error":false,"math_prob_llama3":0.9918706,"pos_list":[0],"im_url_duplicate_count":[null],"WARC_HEADER":"{\"WARC-Type\":\"response\",\"WARC-Date\":\"2019-11-11T20:37:30Z\",\"WARC-Record-ID\":\"<urn:uuid:858f0ad6-7d82-4c30-9e0d-9f36df1bcf5c>\",\"Content-Length\":\"74463\",\"Content-Type\":\"application/http; msgtype=response\",\"WARC-Warcinfo-ID\":\"<urn:uuid:a8b9268b-808c-4df6-9986-f15a17539846>\",\"WARC-Concurrent-To\":\"<urn:uuid:bd16784f-e856-4d96-bd1c-da23f2bf9786>\",\"WARC-IP-Address\":\"39.108.122.106\",\"WARC-Target-URI\":\"https://kexue.fm/archives/7031\",\"WARC-Payload-Digest\":\"sha1:XIJKG2RJSJNRSU4KS7W6DXSMDMOXFJ3N\",\"WARC-Block-Digest\":\"sha1:REN7T6HEGIXJQSIYLJSUQWBNJJMMAL3X\",\"WARC-Identified-Payload-Type\":\"application/xhtml+xml\",\"warc_filename\":\"/cc_download/warc_2019/CC-MAIN-2019-47/CC-MAIN-2019-47_segments_1573496664437.49_warc_CC-MAIN-20191111191704-20191111215704-00269.warc.gz\"}"}