python - 如何使脚本的一部分与 dask 异步？

Question

假设我收到一组文档。我需要对它们进行标记，然后将它们转换成向量以进行进一步的工作。由于我发现 elasticsearch 的分词器比我自己的解决方案效果更好，因此我正在改用它。然而，它的速度要慢得多。然后最终结果预计会以流的形式输入到矢量化器中。

整个过程可以通过生成器链表来完成

def fetch_documents(_cursor):
    with _cursor:
        # a lot of documents expected, may not fit in memory
        _cursor.execute('select ... from ...')

        for doc in _cursor:
            yield doc

def tokenize(documents):
    for doc in documents:
        yield elasticsearch_tokenize_me(doc)

def build_model(documents):
    some_model = SomeModel()

    for doc in documents:
        some_model.add_document(doc)

    return some_model

build_model(tokenize(fetch_documents))

所以这基本上工作正常，但没有利用所有可用的处理能力。由于 dask 在其他相关项目中使用，我尝试适应并得到它(我使用 psycopg2 进行数据库访问)。

from dask import delayed
import psycopg2
import psycopg2.extras
from elasticsearch import Elasticsearch
from elasticsearch.client import IndicesClient

def loader():
    conn = psycopg2.connect()

    cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
    cur.execute('''
                SELECT document, ... FROM ...
                ''')
    return cur

@delayed
def tokenize(partition):
    result = []

    client = IndicesClient(Elasticsearch())

    for row in partition:
        _result = client.analyze(analyzer='standard', text=row['document'])
        result.append(dict(row,
                           tokens=tuple(item['token'] for item in _result['tokens'])))

    return result

@delayed
def build_model(sequence_of_data):
    some_model = SomeModel()

    for item in chain.from_iterable(sequence_of_data):
        some_model.add_document(item)

    return some_model

with loader() as cur:
    partitions = []

    for idx_start in range(0, cur.rowcount, 200):
        partitions.append(delayed(cur.fetchmany)(200))

    tokenized = []
    for partition in partitions:
        tokenized.append(tokenize(partition))

    result = do_something(tokenized)
    result.compute()

代码或多或少都可以工作，除了最后所有文档在输入模型之前都被标记化。虽然这适用于较小的数据集合，但不适用于大量数据集合(由于内存消耗巨大)。我应该只使用普通的 concurrent.futures 来完成这项工作，还是我错误地使用了 dask？

Answer 1

一个简单的解决方案是在计算机上本地加载数据(很难对单个 SQL 查询进行分区)，然后将数据发送到 dask 集群以执行昂贵的标记化步骤。也许是这样的:

rows = cur.execute(''' SELECT document, ... FROM ... ''')

from toolz import partition_all, concat
partitions = partition_all(10000, rows)

from dask.distributed import Executor
e = Executor('scheduler-address:8786')

futures = []

for part in partitions:
    x = e.submit(tokenize, part)
    y = e.submit(process, x)
    futures.append(y)

results = e.gather(futures)
result = list(concat(results))

在此示例中，函数 tokenize 和 process 期望使用并返回元素列表。