Dask DataFrame hiện đã nhanh | bởi Patrick Hoefler | Tháng 5 năm 2024

Cách Dask cho phép xử lý dữ liệu ở quy mô terabyte một cách hiệu quả

Giới thiệu

Dask DataFrame chia tỷ lệ DataFrames của gấu trúc để hoạt động ở quy mô 100GB-100TB.

Trong lịch sử, Dask khá chậm so với các công cụ khác trong lĩnh vực này (như Spark). Do một số cải tiến tập trung vào hiệu suất nên hiện tại nó khá nhanh (nhanh hơn khoảng 20 lần so với trước đây). Việc triển khai mới đã chuyển Dask từ chỗ bị Spark phá hủy trên mọi điểm chuẩn sang thường xuyên vượt trội hơn Spark trên các truy vấn TPC-H một cách đáng kể.

Khối lượng công việc Dask DataFrame gặp khó khăn với nhiều thứ. Hiệu suất và mức sử dụng bộ nhớ thường là những điểm khó khăn, việc xáo trộn không ổn định đối với các tập dữ liệu lớn hơn, khiến việc mở rộng quy mô trở nên khó khăn. Viết mã hiệu quả đòi hỏi phải hiểu quá nhiều về nội bộ của Dask.

Việc triển khai mới đã thay đổi tất cả những điều này. Những thứ không hoạt động đã được viết lại hoàn toàn từ đầu và các triển khai hiện có đã được cải thiện. Điều này đặt Dask DataFrames trên một nền tảng vững chắc cho phép chu kỳ lặp lại nhanh hơn trong tương lai.

Chúng ta sẽ xem xét ba thay đổi nổi bật nhất, đề cập đến cách chúng tác động đến hiệu suất và giúp sử dụng Dask hiệu quả dễ dàng hơn, ngay cả đối với những người dùng mới làm quen với điện toán phân tán. Chúng tôi cũng sẽ thảo luận về kế hoạch cải tiến trong tương lai.

Tôi là thành viên của nhóm cốt lõi của Dask. Tôi là một kỹ sư nguồn mở cho cuộn và đã tham gia thực hiện một số cải tiến được thảo luận trong bài đăng này.

1. Hỗ trợ mũi tên Apache: Kiểu dữ liệu chuỗi hiệu quả

Dask DataFrame bao gồm nhiều DataFrames của gấu trúc. Trong lịch sử, gấu trúc sử dụng NumPy cho dữ liệu số, nhưng các đối tượng Python cho dữ liệu văn bản, điều này không hiệu quả và làm tăng mức sử dụng bộ nhớ. Các hoạt động trên dữ liệu đối tượng cũng giữ GIL, điều này không quan trọng lắm đối với gấu trúc nhưng lại là thảm họa về hiệu suất với một hệ thống music music như Dask.

Bản phát hành pandas 2.0 đã giới thiệu hỗ trợ cho các kiểu dữ liệu Arrow có mục đích chung, vì vậy Dask hiện sử dụng các chuỗi được hỗ trợ bởi PyArrow theo mặc định. đó là nhiều tốt hơn. Chuỗi PyArrow giảm mức sử dụng bộ nhớ tới 80% và mở khóa đa luồng cho các hoạt động chuỗi. Khối lượng công việc trước đây gặp khó khăn với bộ nhớ khả dụng giờ đây vừa vặn thoải mái trong ít không gian hơn và nhanh hơn rất nhiều vì chúng không còn liên tục tràn dữ liệu dư thừa vào đĩa nữa.

Tôi đã viết một bài về điều này điều tra tích hợp Arrow chi tiết hơn nếu bạn muốn tìm hiểu thêm.

2. Tham gia nhanh hơn với thuật toán xáo trộn mới

Xáo trộn là một thành phần thiết yếu của hệ thống phân tán để cho phép sắp xếp, nối và nhóm phức tạp theo các hoạt động. Đây là hoạt động tổng thể, sử dụng nhiều mạng và thường là thành phần đắt tiền nhất trong quy trình làm việc. Chúng tôi đã viết lại hệ thống xáo trộn của Dask, hệ thống này tác động lớn đến hiệu suất tổng thể, đặc biệt là đối với khối lượng công việc phức tạp, sử dụng nhiều dữ liệu.

Hoạt động xáo trộn về bản chất là hoạt động giao tiếp giữa tất cả với tất cả trong đó mọi phân vùng đầu vào phải cung cấp một phần dữ liệu nhỏ cho mọi phân vùng đầu ra. Dask đã sử dụng thuật toán dựa trên nhiệm vụ của riêng mình để giảm thiểu O(n * n) độ phức tạp của nhiệm vụ O(log(n) * n) Ở đâu n là số lượng phân vùng. Đây là sự giảm đáng kể về số lượng tác vụ, nhưng việc mở rộng quy mô phi tuyến tính cuối cùng đã không cho phép Dask xử lý các tập dữ liệu lớn tùy ý.

Dask đã giới thiệu phương pháp xáo trộn P2P (ngang hàng) mới giúp giảm độ phức tạp của nhiệm vụ xuống O(n) tỷ lệ tuyến tính với kích thước của tập dữ liệu và kích thước của cụm. Nó cũng kết hợp tích hợp đĩa hiệu quả cho phép dễ dàng xáo trộn các tập dữ liệu lớn hơn nhiều so với bộ nhớ. Hệ thống mới cực kỳ ổn định và “chỉ hoạt động” trên mọi quy mô dữ liệu.

Một trong những đồng nghiệp của tôi đã viết một bài viết về điều này bao gồm lời giải thích sâu rộng hơn và nhiều chi tiết kỹ thuật.

3. Trình tối ưu hóa

Bản thân Dask lười biếng, có nghĩa là nó đăng ký toàn bộ truy vấn của bạn trước khi thực hiện bất kỳ công việc thực tế nào. Đây là một khái niệm mạnh mẽ cho phép thực hiện nhiều hoạt động tối ưu hóa, nhưng trước đây Dask đã không tận dụng được kiến ​​thức này. Dask cũng đã làm rất tệ khi che giấu sự phức tạp bên trong và khiến người dùng phải tự mình giải quyết những khó khăn của điện toán phân tán và chạy các truy vấn quy mô lớn. Nó làm cho việc viết mã hiệu quả trở nên khó khăn đối với những người không phải là chuyên gia.

Bản phát hành Dask vào tháng 3 bao gồm việc triển khai lại hoàn toàn API DataFrame để hỗ trợ tối ưu hóa truy vấn. Đây là một vấn đề lớn. Công cụ mới xoay quanh trình tối ưu hóa truy vấn viết lại mã của chúng tôi để làm cho mã hiệu quả hơn và phù hợp hơn với các điểm mạnh của Dask. Hãy cùng tìm hiểu một số chiến lược tối ưu hóa, cách chúng giúp mã của chúng ta chạy nhanh hơn và mở rộng quy mô tốt hơn.

Chúng tôi sẽ bắt đầu với một số tối ưu hóa cho mục đích chung hữu ích cho mọi công cụ giống DataFrame trước khi đi sâu vào các kỹ thuật cụ thể hơn được điều chỉnh cho các hệ thống phân tán nói chung và Dask cụ thể hơn.

3.1 Phép chiếu cột

Hầu hết các tập dữ liệu đều có nhiều cột hơn những gì chúng ta thực sự cần. Việc loại bỏ chúng đòi hỏi phải có tầm nhìn xa (“Tôi sẽ cần những cột nào cho truy vấn này? 🤔”) nên hầu hết mọi người không nghĩ đến điều này khi tải dữ liệu. Điều này không tốt cho hiệu suất vì chúng ta mang theo nhiều dữ liệu không cần thiết, làm mọi thứ chậm lại. Phép chiếu Cột loại bỏ các cột ngay khi chúng không còn cần thiết nữa. Đó là một sự tối ưu hóa đơn giản nhưng mang lại lợi ích cao.

Việc triển khai cũ luôn đọc tất cả các cột từ bộ lưu trữ và chỉ loại bỏ các cột nếu chúng tôi chủ động yêu cầu. Chỉ cần hoạt động trên ít dữ liệu hơn là bạn đã đạt được lợi ích lớn về hiệu suất và mức sử dụng bộ nhớ.

Trình tối ưu hóa xem xét truy vấn và tìm ra cột nào cần thiết cho mỗi thao tác. Chúng ta có thể tưởng tượng điều này giống như việc xem xét bước cuối cùng của truy vấn, sau đó thực hiện ngược từng bước đến nguồn dữ liệu và chèn các thao tác thả để loại bỏ các cột không cần thiết.

3.2 Đẩy bộ lọc xuống

Đẩy xuống bộ lọc là một cách tối ưu hóa có mục đích chung khác với cùng mục tiêu như phép chiếu cột: hoạt động trên ít dữ liệu hơn. Việc triển khai cũ chỉ giữ các bộ lọc ở nơi chúng tôi đặt chúng. Việc triển khai mới thực hiện các hoạt động lọc càng sớm càng tốt trong khi vẫn duy trì kết quả tương tự.

Trình tối ưu hóa xác định mọi bộ lọc trong truy vấn của chúng tôi và xem xét thao tác trước đó để xem liệu chúng tôi có thể di chuyển bộ lọc đến gần nguồn dữ liệu hơn hay không. Nó sẽ lặp lại điều này cho đến khi tìm thấy một thao tác không thể chuyển đổi bằng bộ lọc. Việc này khó hơn một chút so với phép chiếu cột vì chúng ta phải đảm bảo rằng các thao tác không làm thay đổi giá trị của DataFrame. Ví dụ: chuyển đổi bộ lọc và thao tác hợp nhất là được (giá trị không thay đổi), nhưng việc chuyển đổi bộ lọc và thao tác thay thế là không hợp lệ, vì giá trị của chúng tôi có thể thay đổi và các hàng trước đây đã được lọc ra bây giờ sẽ không còn nữa , hoặc ngược lại.

Ngoài ra, nếu bộ lọc của chúng tôi đủ mạnh thì chúng tôi có thể loại bỏ các tệp hoàn chỉnh trong bước IO. Đây là trường hợp tốt nhất, trong đó bộ lọc trước đó mang lại sự cải thiện hiệu suất rất lớn và thậm chí yêu cầu đọc ít dữ liệu hơn từ bộ nhớ từ xa.

3.3 Tự động thay đổi kích thước phân vùng

Ngoài việc triển khai các kỹ thuật tối ưu hóa phổ biến được mô tả ở trên, chúng tôi cũng đã cải thiện một điểm yếu chung dành riêng cho các hệ thống phân tán nói chung và người dùng Dask nói riêng: kích thước phân vùng tối ưu.

Dask DataFrames bao gồm nhiều DataFrames gấu trúc nhỏ được gọi là phân vùng. Thông thường, số lượng phân vùng được quyết định cho bạn và người dùng Dask nên “phân vùng lại” theo cách thủ công sau khi giảm hoặc mở rộng dữ liệu của họ (ví dụ: bằng cách thả cột, lọc dữ liệu hoặc mở rộng bằng các phép nối) (xem phần Tài liệu Dask). Nếu không có bước bổ sung này, chi phí hoạt động (thường là nhỏ) từ Dask có thể trở thành nút thắt cổ chai nếu DataFrames của gấu trúc trở nên quá nhỏ, khiến quy trình làm việc của Dask trở nên chậm chạp.

Kiểm soát kích thước phân vùng theo cách thủ công là một nhiệm vụ khó khăn mà chúng tôi, với tư cách là người dùng Dask, không cần phải lo lắng. Nó cũng chậm vì nó yêu cầu chuyển mạng một số phân vùng. Dask DataFrame hiện tự động thực hiện hai việc để trợ giúp khi phân vùng quá nhỏ:

• Giữ kích thước của mỗi phân vùng không đổi, dựa trên tỷ lệ dữ liệu bạn muốn tính toán so với kích thước tệp gốc. Ví dụ: nếu bạn lọc ra 80% tập dữ liệu gốc, Dask sẽ tự động kết hợp các phân vùng nhỏ hơn thành các phân vùng lớn hơn, ít hơn.
• Kết hợp các phân vùng quá nhỏ thành các phân vùng lớn hơn, dựa trên mức tối thiểu tuyệt đối (mặc định là 75 MB). Ví dụ: nếu tập dữ liệu gốc của bạn được chia thành nhiều tệp nhỏ, Dask sẽ tự động kết hợp chúng.

Trình tối ưu hóa sẽ xem xét số lượng cột và kích thước của dữ liệu trong đó. Nó tính toán tỷ lệ được sử dụng để kết hợp nhiều tệp vào một phân vùng.

Bước này hiện bị giới hạn ở các hoạt động IO (như đọc trong tập dữ liệu Parquet), nhưng chúng tôi dự định mở rộng nó sang các hoạt động khác cho phép kết hợp các phân vùng với chi phí thấp.

3.4 Hoạt động hợp nhất và tham gia tầm thường

Các hoạt động hợp nhất và tham gia thường rẻ trên một máy có gấu trúc nhưng đắt tiền trong cài đặt phân tán. Việc hợp nhất dữ liệu trong bộ nhớ dùng chung rất rẻ, trong khi việc hợp nhất dữ liệu trên mạng khá chậm do các thao tác xáo trộn đã được giải thích trước đó.

Đây là một trong những hoạt động tốn kém nhất trong hệ thống phân tán. Việc triển khai cũ đã kích hoạt chuyển mạng của cả hai DataFrame đầu vào cho mọi hoạt động hợp nhất. Điều này đôi khi cần thiết nhưng rất tốn kém.

Trình tối ưu hóa sẽ xác định khi nào cần xáo trộn so với khi nào một phép nối tầm thường là đủ vì dữ liệu đã được căn chỉnh chính xác. Điều này có thể làm cho việc hợp nhất riêng lẻ nhanh hơn rất nhiều. Điều này cũng áp dụng cho các hoạt động khác thường yêu cầu xáo trộn như groupby().apply().

Việc hợp nhất Dask từng không hiệu quả, gây ra thời gian chạy dài. Trình tối ưu hóa sửa lỗi này cho trường hợp tầm thường trong đó các thao tác này xảy ra nối tiếp nhau, nhưng kỹ thuật này vẫn chưa tiến bộ lắm. Vẫn còn rất nhiều tiềm năng để cải thiện.

Trình tối ưu hóa sẽ xem xét biểu thức và chèn các nút xáo trộn khi cần thiết để tránh những xáo trộn không cần thiết.

Các cải tiến được xếp chồng lên nhau như thế nào so với việc triển khai cũ?

Dask bây giờ nhanh hơn 20 lần so với trước đây. Cải tiến này áp dụng cho toàn bộ API DataFrame (không chỉ các thành phần riêng biệt), không có hồi quy hiệu suất nào được biết đến. Dask hiện chạy khối lượng công việc mà trước đây không thể hoàn thành trong khung thời gian chấp nhận được. Việc tăng hiệu suất này là do có nhiều cải tiến được xếp chồng lên nhau. Vấn đề không phải là làm một việc đặc biệt tốt mà là không làm gì đặc biệt kém.

Hiệu suất, mặc dù là cải tiến hấp dẫn nhất, nhưng không phải là điều duy nhất trở nên tốt hơn. Trình tối ưu hóa che giấu rất nhiều sự phức tạp đối với người dùng và làm cho việc chuyển đổi từ pandas sang Dask trở nên dễ dàng hơn rất nhiều vì giờ đây việc viết mã hoạt động kém đã khó khăn hơn nhiều. Toàn bộ hệ thống mạnh mẽ hơn.

Kiến trúc mới của API cũng dễ làm việc hơn rất nhiều. Việc triển khai cũ đã làm rò rỉ rất nhiều vấn đề phức tạp bên trong vào quá trình triển khai API cấp cao, khiến các thay đổi trở nên cồng kềnh. Những cải tiến gần như không đáng kể để thêm vào ngay bây giờ.

Điều gì sẽ đến?

Dask DataFrame đã thay đổi rất nhiều trong 18 tháng qua. API cũ thường khó làm việc và gặp khó khăn trong việc mở rộng quy mô. Việc triển khai mới đã loại bỏ những thứ không hoạt động và cải thiện các hoạt động triển khai hiện có. Công việc nặng nhọc hiện đã hoàn tất, cho phép các chu kỳ lặp lại nhanh hơn để cải thiện hiện trạng. Những cải tiến gia tăng giờ đây không còn quá khó để thêm vào.

Một số điều nằm trong lộ trình trước mắt:

• Tự động phân vùng lại: điều này được triển khai một phần, nhưng có nhiều khả năng hơn để chọn kích thước phân vùng hiệu quả hơn trong quá trình tối ưu hóa.
• Tham gia nhanh hơn: vẫn còn rất nhiều điều chỉnh cần thực hiện ở đây. Ví dụ: chúng tôi có PR trong chuyến bay với mức cải thiện 30–40%.
• Tham gia sắp xếp lại: chúng tôi chưa làm điều này nhưng nó đã nằm trong lộ trình trước mắt

Bài viết này tập trung vào một số cải tiến đối với Dask DataFrame và kết quả là nó nhanh hơn và đáng tin cậy hơn bao nhiêu. Nếu đang chọn giữa Dask và các công cụ DataFrame phổ biến khác, bạn cũng có thể cân nhắc:

Cảm ơn bạn đã đọc. Hãy liên hệ để chia sẻ suy nghĩ và phản hồi của bạn.