การใช้ process และ thread

Parallel Programming with Dask in Python

James Fulton

Climate Informatics Researcher

Dask default scheduler

Threads

Dask arrays
Dask DataFrames
Delayed pipelines ที่สร้างด้วย dask.delayed()

Processes

Dask bags

การเลือก scheduler

# Use default
result = x.compute()

result = dask.compute(x)


# Use threads
result = x.compute(scheduler='threads')

result = dask.compute(x, scheduler='threads')


# Use processes
result = x.compute(scheduler='processes')

result = dask.compute(x, scheduler='processes')

สรุป - threads และ processes

Threads

เริ่มต้นได้รวดเร็วมาก
ไม่ต้องโอนถ่ายข้อมูลไปให้
ถูกจำกัดด้วย GIL ซึ่งอนุญาตให้อ่านโค้ดได้ทีละ thread

Processes

ใช้เวลาในการตั้งค่า
โอนถ่ายข้อมูลได้ช้า
แต่ละ process มี GIL ของตัวเอง จึงไม่ต้องผลัดกันอ่านโค้ด

การสร้าง local cluster

from dask.distributed import LocalCluster

cluster = LocalCluster(
    processes=True, 
    n_workers=2,
    threads_per_worker=2
)

print(cluster)

LocalCluster(..., workers=2, threads=4, memory=31.38 GiB)

การสร้าง local cluster

from dask.distributed import LocalCluster

cluster = LocalCluster(
    processes=False, 
    n_workers=2,
    threads_per_worker=2
)

print(cluster)

LocalCluster(..., workers=2, threads=4, memory=31.38 GiB)

local cluster แบบง่าย

cluster = LocalCluster(processes=True)

print(cluster)

LocalCluster(..., workers=4 threads=8, memory=31.38 GiB)

cluster = LocalCluster(processes=False)

print(cluster)

LocalCluster(..., workers=1 threads=8, memory=31.38 GiB)

การสร้าง client

from dask.distributed import Client, LocalCluster
cluster = LocalCluster(
    processes=True, 
    n_workers=4,
    threads_per_worker=2
)

client = Client(cluster)
print(client)

<Client: 'tcp://127.0.0.1:61391' processes=4 threads=8, memory=31.38 GiB>

การสร้าง client อย่างรวดเร็ว

สร้าง cluster ก่อน แล้วส่งเข้า client

cluster = LocalCluster(
    processes=True, 
    n_workers=4,
    threads_per_worker=2
)

client = Client(cluster)

print(client)

<Client: ... processes=4 threads=8, ...>

สร้าง client ซึ่งจะสร้าง cluster ของตัวเองโดยอัตโนมัติ

client = Client(
    processes=True, 
    n_workers=4,
    threads_per_worker=2
)



print(client)

<Client: ... processes=4 threads=8, ...>

การใช้งาน cluster

client = Client(processes=True)

# Default uses the client
result = x.compute()


# Can still change to other schedulers
result = x.compute(scheduler='threads')


# Can explicitly use client
result = client.compute(x)

cluster ประเภทอื่น

LocalCluster() - cluster บนเครื่องของคุณ
cluster ประเภทอื่นกระจายการประมวลผลไปยังเครื่องหลายเครื่อง

มาฝึกกันเถอะ!

Parallel Programming with Dask in Python