Tăng tốc quy trình khoa học dữ liệu với Dask: Hướng dẫn toàn diện

Khi Wes McKinney bắt đầu viết pandas, ông có một nguyên tắc kinh nghiệm: để pandas hoạt động tối ưu, dung lượng RAM của máy phải lớn hơn kích thước tập dữ liệu tương ứng từ 5–10 lần. Quy tắc này khá dễ tuân theo vào khoảng năm 2010, nhưng bây giờ đã là 2023.

Đến năm 2020, các tập dữ liệu thực tế đã tăng đến kích thước có thể dễ dàng làm sập máy tính xách tay và máy tính thông thường. Dự đoán vấn đề này từ sớm, một giải pháp đã được phát hành vào năm 2015.

Dask là một thư viện mã nguồn mở được phát triển bởi nhóm tạo ra Anaconda nhằm giải quyết thách thức về tính mở rộng và hiệu năng khi tính toán trên các tập dữ liệu lớn vượt quá khả năng bộ nhớ của một máy đơn lẻ.

Hướng dẫn này cung cấp phần giới thiệu toàn diện về Dask và các tính năng cốt lõi của nó, bao gồm các giao diện cho DataFrame, Array và Bag.

Cài đặt Dask

Giống như các thư viện khác, bạn có thể cài đặt Dask theo ba cách: Conda, Pip và từ mã nguồn.

Vì đây là bài viết nhập môn về Dask, chúng tôi sẽ không đề cập đến phương pháp cài đặt cuối cùng, vì nó dành cho người bảo trì.

Nếu bạn dùng Anaconda, Dask đã được bao gồm trong bản cài đặt mặc định (điều này thể hiện mức độ phổ biến của thư viện). Nếu muốn cài lại hoặc nâng cấp, bạn có thể dùng lệnh install:

conda install dask

Phương án dùng PIP như sau:

pip install "dask[complete]"

Thêm phần mở rộng [complete] cũng sẽ cài đặt các phụ thuộc cần thiết của Dask, loại bỏ nhu cầu cài thủ công NumPy, Pandas và Tornado.

Bạn có thể kiểm tra việc cài đặt có thành công hay không bằng cách xem phiên bản thư viện:

import dask

dask.__version__

Kết quả:

'2023.5.0'

Phần lớn công việc với Dask sẽ xoay quanh ba giao diện: Dask DataFrame, Array và Bag. Hãy import chúng cùng với numpy và pandas để dùng cho phần còn lại của bài viết:

import dask.array as da
import dask.bag as db
import dask.dataframe as dd
import numpy as np
import pandas as pd

Các khái niệm cơ bản của Dask

Ở mức khái quát, bạn có thể coi Dask như một lớp bọc mở rộng khả năng của các công cụ truyền thống như pandas, NumPy và Spark để xử lý các tập dữ liệu lớn hơn bộ nhớ.

Khi đối mặt với các đối tượng lớn như mảng (vector) hoặc ma trận (dataframe) vượt quá bộ nhớ, Dask sẽ chia nhỏ chúng thành các khối, còn gọi là partition.

Ví dụ, hãy xét mảng 12 số ngẫu nhiên trong cả NumPy và Dask:

narr = np.random.rand(12)

narr

array([0.9261154 , 0.87774082, 0.87078873, 0.22309476, 0.24575174,
      0.04182393, 0.31476305, 0.04599283, 0.62354124, 0.97597454,
      0.23923457, 0.81201211])

darr = da.from_array(narr, chunks=3)
darr

Hình trên cho thấy mảng Dask chứa bốn khối vì chúng ta đặt chunks là 3. Bên trong, mỗi khối bản thân nó là một mảng NumPy.
Giờ hãy xét một ví dụ lớn hơn nhiều. Chúng ta sẽ tạo hai mảng 10k x 100k (1 tỷ phần tử) và thực hiện phép nhân theo phần tử ở cả hai thư viện đồng thời đo hiệu năng:
# Create the NumPy arrays
arr1 = np.random.rand(10_000, 100_000)
arr2 = np.random.rand(10_000, 100_000)

# Create the Dask arrays
darr1 = da.from_array(arr1, chunks=(1_000, 10_000))
darr2 = da.from_array(arr2, chunks=(1_000, 10_000))
%%time

result_np = np.multiply(arr1, arr2)
Wall time: 3.19 s
%%time

result_dask = da.multiply(darr1, darr2)
Wall time: 94.8 ms
Đoạn mã trên minh họa phép nhân theo phần tử của hai mảng lớn bằng cả NumPy và Dask. Như kết quả cho thấy, Dask nhanh hơn NumPy khoảng 34 lần cho phép tính này. Lợi ích hiệu năng còn rõ hơn khi độ phức tạp phép tính và kích thước mảng tăng lên.
Dask cũng áp dụng cách tiếp cận chia khối và phân phối các khối này lên toàn bộ lõi có sẵn trên máy của bạn đối với các đối tượng khác.
Dask DataFrame
Để thấy hết lợi ích của Dask, chúng ta cần một tập dữ liệu lớn, tốt nhất là trên 1 GB. Tuy nhiên, tải xuống một tập dữ liệu như vậy để theo dõi trong hướng dẫn có thể không tối ưu. Thay vào đó, bạn có thể dùng đoạn script này, tạo tập dữ liệu tổng hợp với 10 triệu dòng, 10 đặc trưng số và 10 đặc trưng phân loại.
Vui lòng đảm bảo máy của bạn có ít nhất 12 GB RAM để chạy script.
Khi đã có tệp large_dataset.csv trong không gian làm việc, bạn có thể nạp nó bằng hàm read_csv từ giao diện Dask DataFrame (dd):
import dask.dataframe as dd

dask_df = dd.read_csv("data/large_dataset.csv")

dask_df.head()

Mặc dù tệp lớn, bạn sẽ thấy kết quả được lấy gần như tức thì. Với các tệp còn lớn hơn, bạn có thể chỉ định tham số `blocksize`, tham số này quyết định số byte để chia nhỏ tệp.
Tương tự như cách Dask Array chứa các khối mảng NumPy nhỏ, Dask được thiết kế để xử lý nhiều DataFrame Pandas nhỏ được sắp xếp theo chỉ mục hàng.

Như bạn có thể đoán từ hàm read_csv, phần lớn cú pháp và chức năng thường dùng của API pandas được giữ nguyên trong Dask. Các khối mã sau đây hẳn đã quen thuộc với bạn khi làm việc với pandas.
Chọn cột và các phép toán theo phần tử
Trong ví dụ này, chúng ta thực hiện một số phép toán cột khá đơn giản trên Dask DataFrame có tên dask_df. Chúng ta cộng các giá trị từ cột Numeric_0 với kết quả của phép nhân các giá trị từ Numeric_9 và Numeric_3. Chúng ta lưu kết quả vào biến result.
result = (
   dask_df["Numeric_0"] + dask_df["Numeric_9"] * dask_df["Numeric_3"]
)

result.compute().head()


0    1.301960
1    1.190679
2    1.100955
3    0.758272
4    0.926729
dtype: float64
Như đã đề cập, Dask hơi khác các công cụ tính toán truyền thống ở chỗ nó không thực thi ngay các phép toán này. Thay vào đó, nó tạo ra một kiểu "kế hoạch" gọi là đồ thị tác vụ (task graph) để thực hiện các phép toán sau. Cách tiếp cận này cho phép Dask tối ưu hóa và song song hóa khi cần. Hàm compute() kích hoạt Dask thực hiện các phép tính, còn head() chỉ hiển thị vài dòng đầu của kết quả.
Lọc có điều kiện
Bây giờ, hãy xem cách Dask lọc dữ liệu. Chúng ta chọn các hàng trong DataFrame mà giá trị ở cột "Categorical_5" là "A".
Quá trình lọc này tương tự như cách bạn làm trong pandas, nhưng có một điểm khác - Dask thực hiện theo kiểu lười. Nó chuẩn bị đồ thị tác vụ cho phép toán này nhưng chờ đến khi gọi compute() mới thực thi. Khi chạy head(), chúng ta sẽ thấy vài hàng đầu tiên của DataFrame đã lọc.
dask_df[dask_df["Categorical_5"] == "A"].compute().head()

Thống kê tóm tắt phổ biến
Tiếp theo, chúng ta sẽ tạo một số thống kê tóm tắt phổ biến bằng hàm describe() của Dask.
Hàm này cung cấp một loạt thống kê mô tả cho DataFrame của chúng ta, gồm trung bình, độ lệch chuẩn, giá trị nhỏ nhất, lớn nhất, v.v. Như các ví dụ trước, Dask chuẩn bị đồ thị tác vụ khi gọi describe(), nhưng chờ đến khi gọi compute() mới thực thi.
dask_df.describe().compute()

dask_df["Categorical_3"].value_counts().compute().head()
Categorical_3
O    386038
C    385804
A    385493
P    385490
K    385116
Name: count, dtype: int64
Chúng ta cũng dùng value_counts() để đếm số lần xuất hiện của mỗi giá trị duy nhất trong cột "Categorical_3". Ta kích hoạt phép tính bằng compute(), và head() hiển thị các giá trị phổ biến nhất.
Groupby
Cuối cùng, hãy dùng hàm groupby() để nhóm dữ liệu theo các giá trị trong cột "Categorical_8". Sau đó chọn cột "Numeric_7" và tính trung bình cho mỗi nhóm.
Cách làm này tương tự như khi bạn dùng ‘groupby()’ trong pandas, nhưng như bạn đoán, Dask thực hiện một cách lười. Chúng ta kích hoạt phép tính bằng compute(), và head() hiển thị giá trị trung bình của cột "Numeric_7" cho vài nhóm đầu tiên.
dask_df.groupby("Categorical_8")["Numeric_7"].mean().compute().head()


Categorical_8
A    0.498497
B    0.499767
C    0.500622
D    0.500307
E    0.499530
Name: Numeric_7, dtype: float64

Xem thêm phần này trong hướng dẫn sử dụng Dask về các điểm tương đồng khác giữa pandas và Dask.
Đánh giá lười (Lazy evaluation)
Bây giờ, hãy tìm hiểu việc dùng hàm compute ở cuối mỗi khối mã.
Dask đánh giá các khối mã theo chế độ lười so với chế độ háo hức (eager) của Pandas, nơi kết quả được trả về ngay lập tức.
Để so sánh với việc nấu ăn, đánh giá lười giống như chuẩn bị sẵn nguyên liệu và thái rau củ trước, nhưng chỉ trộn và nấu khi cần. Hàm compute chính là để làm việc đó.
Ngược lại, đánh giá háo hức giống như đưa nguyên liệu vào nấu ngay khi chúng sẵn sàng. Cách tiếp cận này đảm bảo mọi thứ được phục vụ cùng lúc.
Đánh giá lười là chìa khóa cho hiệu năng xuất sắc của Dask vì nó mang lại:

Giảm tính toán. Biểu thức chỉ được đánh giá khi cần (khi gọi compute), tránh tạo ra các kết quả trung gian không cần thiết có thể không được dùng trong kết quả cuối.
Phân bổ tài nguyên tối ưu. Đánh giá lười tránh việc cấp phát bộ nhớ hoặc năng lực xử lý cho các kết quả trung gian có thể không cần.
Hỗ trợ tập dữ liệu lớn. Phương pháp này xử lý dữ liệu theo thời gian thực hoặc thành các khối nhỏ hơn, giúp sử dụng hiệu quả tài nguyên bộ nhớ.

Khi kết quả của compute được trả về, chúng sẽ là Series/DataFrame của Pandas hoặc mảng NumPy thay vì Dask DataFrame gốc.
>>> type(dask_df)
dask.dataframe.core.DataFrame


>>> type(
   dask_df[["Numeric_5", "Numeric_6", "Numeric_7"]].mean().compute()
)

pandas.core.series.Series

Lý do là hầu hết các phép biến đổi dữ liệu chỉ trả về một tập con của dataframe ban đầu, chiếm ít dung lượng hơn nhiều. Vì vậy, sẽ không cần đến khả năng song song của Dask nữa, và bạn có thể tiếp tục phần còn lại của quy trình làm việc trong pandas hoặc NumPy.
Dask Bag và Dask Delayed cho dữ liệu phi cấu trúc
Dask Bag và Dask Delayed là hai thành phần của thư viện Dask cung cấp công cụ mạnh mẽ để làm việc với dữ liệu phi cấu trúc hoặc bán cấu trúc và hỗ trợ đánh giá lười.
Nếu trước đây dữ liệu dạng bảng là phổ biến nhất, thì ngày nay các tập dữ liệu thường bao gồm tệp phi cấu trúc như ảnh, tệp văn bản, video và âm thanh. Dask Bag cung cấp chức năng và API để xử lý các tệp phi cấu trúc như vậy theo cách song song và có khả năng mở rộng.
Ví dụ, hãy xét minh họa đơn giản sau:
import dask.bag as db

# Create a Dask Bag from a list of strings
b = db.from_sequence(["apple", "banana", "orange", "grape", "kiwi"])

# Filter the strings that start with the letter 'a'
filtered_strings = b.filter(lambda x: x.startswith("a"))

# Map a function to convert each string to uppercase
uppercase_strings = filtered_strings.map(lambda x: x.upper())

# Compute the result as a list
result = uppercase_strings.compute()

print(result)
['APPLE']
Trong ví dụ này, chúng ta tạo một Dask Bag b từ danh sách chuỗi. Sau đó, áp dụng các thao tác trên Bag để lọc những chuỗi bắt đầu bằng chữ 'a' và chuyển chúng thành chữ hoa bằng các hàm filter() và map(). Cuối cùng, chúng ta tính kết quả thành danh sách bằng phương thức compute() và in ra.
Giờ hãy tưởng tượng bạn có thể thực hiện các thao tác phức tạp hơn trên hàng tỷ chuỗi tương tự được lưu trong một tệp văn bản. Nếu không có đánh giá lười và song song mà Dask Bag cung cấp, bạn sẽ gặp thách thức lớn. (Đọc thêm về Bag trong tài liệu Dask).
Về Dask Delayed, nó cung cấp sự linh hoạt hơn nữa và đưa đánh giá lười cùng song song hóa vào nhiều tình huống khác. Với Dask Delayed, bạn có thể chuyển bất kỳ hàm Python gốc nào thành một đối tượng lười bằng decorator @dask.delayed.
Đây là một ví dụ đơn giản:
%%time

import time
import dask


@dask.delayed
def process_data(x):
   # Simulate some computation
   time.sleep(1)
   return x**2


# Generate a list of inputs
inputs = range(1000)

# Apply the delayed function to each input
results = [process_data(x) for x in inputs]

# Compute the results in parallel
computed_results = dask.compute(*results)



Wall time: 42.1 s

Trong ví dụ này, chúng ta định nghĩa hàm process_data được trang trí bằng @dask.delayed. Hàm mô phỏng công việc tính toán bằng cách ngủ 1 giây rồi trả về bình phương của giá trị đầu vào.
Nếu không có song song, thực hiện phép tính này trên 1000 đầu vào sẽ mất hơn 1000 giây. Tuy nhiên, với Dask Delayed và thực thi song song, phép tính chỉ mất khoảng 42,1 giây.
Ví dụ này cho thấy sức mạnh của song song hóa trong việc rút ngắn thời gian tính toán bằng cách phân phối tải công việc hiệu quả lên nhiều lõi hoặc worker.
Đó chính là ý nghĩa của song song hóa.
Kết luận và tài nguyên tham khảo
Dask là một trong những thư viện nền tảng trong hệ sinh thái dữ liệu. Nó mở rộng chức năng của hầu hết các thư viện được yêu thích như NumPy, Pandas và Spark, cho phép xử lý liền mạch các tập dữ liệu lớn hơn bộ nhớ.
Với Dask Bag và Dask Delayed, nó mang song song hóa và đánh giá lười đến các kịch bản không truyền thống như làm việc với dữ liệu phi cấu trúc hoặc với đối tượng Python gốc.
Để nắm các chi tiết nhỏ, hãy đọc kỹ tài liệu Dask.
Nếu bạn đang tìm một nguồn học toàn diện để làm chủ Dask, hãy xem khóa học của chúng tôi,  Parallel Programming With Dask in Python.