Dask #3. 대스크 compute()와 persist()

1. 복잡한 DAG 생성

 

실무에서 사용하는 복잡한 로직과 비교할 수 없지만

compute()와 persist()를 비교하기 위해서 복잡한 DAG를 생성합니다.

 

import dask.delayed as delayed
from dask.diagnostics import ProgressBar

def add_two(x) :
    return x + 2

def sum_two_numbers(x, y) :
    return x + y

def multiply_four(x) :
    return x * 4

data = [1, 5, 8, 10]

step1 = [delayed(add_two)(i) for i in data]
step2 = [delayed(multiply_four)(j) for j in step1]
total = delayed(sum)(step2)
total.visualize()

 

 

단순한 덧셈 연산에 곱하기 연산을 추가한 DAG를 생성하고도 아직 단순한 것 같습니다.

위에서 생성한 DAG에 다른 계층을 추가해 보겠습니다.

 

data2 = [delayed(sum_two_numbers)(k, total) for k in data]
total2 = delayed(sum)(data2)
total2.visualize()

 

 

조금은 더 복잡한 DAG가 완성되었습니다.

대스크는 이러한 연산의 흐름을 생성하고 저장합니다. 다만 아직 연산이 시작된 것은 아닙니다.

이제 생성된 DAG로 compute와 persist를 비교해 보도록 하겠습니다.

 

2. compute() vs persist()

 

위에서 생성된 total2를 통해 compute() 메서드로 연산을 진행합니다.

 

with ProgressBar() :
    print(total2.compute())
total2.visualize()

 

 

compute()를 사용하면 536의 연산 결과를 도출합니다. 하지만 DAG는 유지가 됩니다.

지연 객체에서 compute 메서드를 호출할 때마다 536이라는 결과를 도출하기 위해서 전체 DAG를 단계적으로 실행합니다.

만약에 실무에서 반복해서 사용되는 결과라고 가정했을 때, 해당 연산을 계속해서 반복하게 됩니다.

이를 효율적으로 사용하기 위해서 persist를 사용합니다.

 

with ProgressBar() :
    persisted_data = total2.persist()
    print(persisted_data)
persisted_data.visualize()

 

persist 사용 후 DAG

 

위 코드처럼 persist 메서드를 사용하면 결과를 도출하지는 않지만 지연 객체를 생성하는데, DAG가 단일 결과를 얻었습니다.

자주 사용하게 되는 연산이라면 이처럼 중간 결과를 유지하는(persist)것이 효율적일 것입니다.

 

data3 = [delayed(sum_two_numbers)(l, persisted_data) for l in data]
total3 = delayed(sum)(data3)
total3.visualize()

 

유지(persist)한 DAG에 다른 DAG 연결하기

 

자주 사용되는 로직을 중간 저장 후에 위 코드처럼 새로운 DAG를 추가할 수 있습니다.