Блог

Все еще пытаюсь понять, как Dask и Kubernetes могут работать с другими контейнерами, поэтому надеялся, что кто-нибудь вообще сможет сказать, может ли сработать следующее? В частности, я не совсем понимаю, может ли/как Dask distributed распространять данные в буферные модули в кластере, в котором работает другой контейнер, который анализирует эти данные и возвращает их обратно в dask для последующих функций. Здесь «другие» контейнеры представляют собой скомпилированные программы, которые преобразуют данные.

Что-то сродни нижеприведенному:

 import dask
from dask import delayed, compute
from dask_kubernetes import KubeCluster
cluster = KubeCluster('worker-spec.yml')
cluster.scale(10)
client = Client(cluster)

@delayed
def run_transformer(raw_data):
    transformed_data = run_transformer_container(raw_data)
    return transformed_data

@delayed
def upload_to_s3(transformed_data):
    success = True
    [...]
    return success

raw_data = [string1, string2, ..., stringN]

output = []
for x in raw_data:
    f = run_transformer(x)
    g = upload_to_s3(f)
    output.append(g)

transformed_data = compute(output)
 

Где ДАСК с задержкой обрабатывает распределения Н задачи до 10 рабочих узлов, каждый из которых проходит raw_data содержимое (вероятно, или, возможно, маринованные объект) в очереди до стручок на то, что рабочий узел, содержащий контейнер, который будет поглощать и преобразовывать данные и вернуть анализируемых данных (через неуказанный run_transformer_container функции, однако, что будет работать) перед загрузкой в S3.