Spark Đọc dữ liệu JDBC đồng thời

Bạn đã bao giờ trải qua quá trình triển khai Spark trong dự án của mình chưa, xác định số phân vùng xáo trộn tối ưu, phân bổ bộ nhớ cho các phiên bản trình điều khiển và bộ thực thi, số lõi bộ thực thi và tất cả những thứ thú vị đó chỉ để đọc dữ liệu từ nguồn JDBC như thế này ?

jdbcDF = spark.read \
    .format("jdbc") \
    .option("driver", "org.postgresql.Driver") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("user", os.environ['user']) \
    .option("password", os.environ['pass']) \
    .option("query", query) \
    .load()

Có vấn đề gì với đoạn mã trên? Chà…nó rất chậm :) Nếu bạn thực thi một truy vấn SQL trong Spark như trong ví dụ trên, bạn sẽ chỉ sử dụng một luồng (bạn có thể thấy trong Giao diện người dùng Spark chỉ có một tác vụ chạy trong khi thực thi mã) và khung dữ liệu được tạo sẽ có một phân vùng.

Hầu hết chúng ta sử dụng Spark trong các đường dẫn ETL/ELT để xử lý song song, do đó, việc đọc dữ liệu từ nguồn JDBC chỉ bằng một luồng có lẽ không phải là điều chúng ta muốn. Chưa kể chúng ta sẽ phải chia lại dataframe sau khi thực hiện truy vấn vì dữ liệu lúc này được lưu trữ trong 1 partition. Vậy làm thế nào để chúng ta thực sự đọc dữ liệu đồng thời trong Spark?

# pass the original query to dbtable as a subquery 
final_query = f'({query}) as q'

# depends on your usecase, how many executor cores you have in the cluster
# what do you plan to do with the dataframe downstream, data size etc
partitionsCount = 100

# I'm hardcoding the date values here, but I'll show below how to identify
# these values at run time if you don't know them upfront
min_date = '2020-01-01'
max_date = '2022-12-31'

jdbcDF = spark.read \
    .format("jdbc") \
    .option("driver", "org.postgresql.Driver") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("user", os.environ['user']) \
    .option("password", os.environ['pass']) \
    .option("dbtable", final_query) \
    .option("numPartitions", partitionsCount) \
    .option("partitionColumn", "date") \
    .option("lowerBound", f"{min_date}") \
    .option("upperBound", f"{max_date}") \
    .load()

1. dbtable:
   Dựa trên tài liệu chính thức từ Spark, nếu chúng tôi muốn đọc dữ liệu đồng thời và phân vùng dữ liệu dựa trên nhu cầu của mình, chúng tôi không còn có thể sử dụng tùy chọn truy vấn và thay vào đó phải chuyển truy vấn SQL của chúng tôi thông qua tùy chọn dbtable . Sự khác biệt duy nhất giữa hai tùy chọn này (ngoài tên) là truy vấn bây giờ phải được đặt trong dấu ngoặc đơn và có bí danh — về cơ bản chuyển truy vấn ban đầu dưới dạng truy vấn phụ.
2. numPartitions:
   Số lượng phân vùng tối đa có thể được sử dụng để đọc và ghi bảng song song. Điều này cũng xác định số lượng kết nối JDBC đồng thời tối đa. Nếu số lượng phân vùng cần ghi vượt quá giới hạn này, chúng tôi sẽ giảm nó xuống giới hạn này bằng cách gọicolesce(numPartitions) trước khi ghi. ( nguồn ) Tôi đã đặt giá trị của mình thành 100 trong cụm EMR, nhưng một cách thực hành tốt là đặt số phân vùng từ 1 đến 4 lần số lõi. Tuy nhiên, đó không phải là một con số cố định, vì vậy hãy luôn thử nghiệm với các giá trị khác nhau để xem giá trị nào phù hợp với bạn.
3. partitionColumn:
   Đây có lẽ là tham số quan trọng nhất và là tham số sẽ tạo ra sự khác biệt lớn nhất trong thời gian chạy truy vấn của bạn. Một số điều cần lưu ý:
   - cột được đề cập phải là số, ngày tháng hoặc dấu thời gian
   - để có hiệu suất tốt nhất có thể, các giá trị cột phải được phân bổ đồng đều nhất có thể và có lực lượng cao. Chúng tôi muốn tránh các cột bị sai lệch dữ liệu — tôi sẽ mở rộng vấn đề này bên dưới.
   - nếu bạn có nhiều cột với các thuộc tính trên, hãy chọn cột được lập chỉ mục.
4. Lowerbound, Upperbound :
   Lưu ý rằng LowerBound và UpperBound chỉ được sử dụng để quyết định bước phân vùng, không phải để lọc các hàng trong bảng. Vì vậy, tất cả các hàng trong bảng sẽ được phân vùng và trả về. ( nguồn ) Bên dưới, bạn sẽ thấy cách Spark tạo từng phân vùng. Đây là lý do tại sao bạn không muốn một cột có số lượng thẻ thấp làm cột phân vùng của mình, nhưng tôi sẽ đưa ra một ví dụ bên dưới phòng trường hợp hiện tại nó không hợp lý.

Hãy tưởng tượng rằng cột bạn muốn sử dụng để phân vùng chỉ có các giá trị 0 và 1 — thực sự là một ví dụ cực đoan, nhưng lại là một kịch bản phổ biến trong Kỹ thuật dữ liệu nơi các giá trị boolean Đúng và Sai được chuyển đổi thành giá trị số.

Vì bạn chỉ có 0 và 1, nên bạn sử dụng 0 làm giới hạn dưới và 1 làm giới hạn trên. Dựa trên cách các phân vùng được tạo - xem hình trên - tất cả các hàng có giá trị 0 sẽ được đẩy đến phân vùng đầu tiên (mệnh đề WHERE của phân vùng đầu tiên là mệnh đề duy nhất trong đó 0 không được lọc ra) và tất cả các hàng với giá trị 1 sẽ được đẩy đến phân vùng cuối cùng ( mệnh đề WHERE của phân vùng cuối cùng là mệnh đề duy nhất trong đó 1 không được lọc ra) .

Vì vậy, bất kể bạn tạo bao nhiêu phân vùng, chỉ có hai phân vùng sẽ được điền và chỉ có hai lõi thực thi sẽ thực hiện tất cả công việc, những phân vùng khác sẽ ở trạng thái không hoạt động nếu không có công việc nào khác được lên lịch cho chúng. Do đó, hiệu suất truy vấn của bạn sẽ bị ảnh hưởng.

Độ lệch dữ liệu

Hãy tưởng tượng chúng ta có một tập dữ liệu với 20 triệu hàng và 30 phân vùng, giới hạn dưới và trên là 2020–01–01 và 2022–12–31. Với mục đích của bài viết này, tôi sẽ sử dụng Phiên Spark với 30 lõi thực thi (có nghĩa là chúng tôi có thể chạy tối đa 30 tác vụ cùng một lúc). Hãy nhớ rằng Spark chỉ định một tác vụ cho mỗi phân vùng, vì vậy mỗi phân vùng sẽ được xử lý bởi một lõi thực thi.

Dữ liệu có giá trị mỗi năm sẽ được chia thành 10 phân vùng và coi như 1 nhiệm vụ sẽ được giao cho 1 phân vùng, mỗi nhiệm vụ được giao cho năm 2020 và 2021 sẽ chịu trách nhiệm xử lý 100 nghìn hàng, nhưng mỗi nhiệm vụ được giao cho năm 2022 sẽ phải xử lý 1,8 triệu hàng. Đó là mức tăng 1700% đối với dữ liệu phải được xử lý bởi một tác vụ.

Kết quả là 20 tác vụ đầu tiên sẽ hoàn thành xử lý phân vùng được chỉ định của chúng ngay lập tức và sau đó 20 lõi thực thi được gán cho các tác vụ đã hoàn thành sẽ không hoạt động (nếu không có công việc nào khác mà các lõi thực thi nhàn rỗi có thể được sử dụng) cho đến tác vụ cuối cùng 10 nhiệm vụ hoàn thành xử lý thêm 1700% hồ sơ. Điều này gây lãng phí tài nguyên và tăng chi phí vận hành, đặc biệt nếu bạn sử dụng Spark trong cụm EMR, Glue Job hoặc Databricks.

Làm thế nào để chúng tôi khắc phục điều đó? Giải pháp đơn giản nhất: sử dụng một cột khác để phân vùng. Nhưng nếu việc sử dụng cột này là quan trọng, chúng ta có thể sử dụng hai truy vấn và hai khung dữ liệu thay vì một. Bạn vẫn sẽ gặp sự cố với dữ liệu bị sai lệch ở hạ lưu (sự cố lặp lại trong các hệ thống tính toán song song) nhưng chúng tôi giải quyết từng sự cố một.

Truy vấn đầu tiên sẽ tải hai năm đầu tiên (30 lõi thực thi sẽ được phân chia từ năm 2020 đến năm 2021), sau đó truy vấn thứ hai sẽ tải năm cuối cùng (vì vậy tất cả 30 lõi thực thi hiện sẽ được sử dụng cho năm 2022). Do đó, một nhiệm vụ cụ thể liên quan đến năm 2022 giờ đây sẽ chịu trách nhiệm xử lý 600 nghìn hàng thay vì 1,8 triệu hàng. Đó là mức giảm 66% trong các hàng phải được xử lý bởi một lõi thực thi duy nhất so với triển khai ban đầu và trên hết, giờ đây chúng tôi sử dụng tất cả các lõi thực thi từ Phiên Spark của chúng tôi thay vì để chúng ở trạng thái chờ.

Điều gì xảy ra nếu tôi không có bất kỳ cột số, ngày tháng hoặc dấu thời gian nào hoặc cột có dữ liệu bị sai lệch?

Không thành vấn đề. Với tư cách là Kỹ sư dữ liệu, việc sử dụng CTE và Hàm cửa sổ để tạo cột số thông qua row_number()sẽ là một miếng bánh. Một số chi phí sẽ được đưa vào khi cột mới được tạo, nhưng ngay cả với chi phí này, truy vấn cuối cùng sẽ được thực thi nhanh hơn bằng cách sử dụng cột mới được tạo thay vì cố tải tất cả dữ liệu trong một phân vùng bằng một lõi thực thi.

with cte as
(
    SELECT f1, f2, f3
    FROM table
)
select *, row_number() over (order by f1) as rn from cte

original_query = "SELECT f1, f2, f3 FROM table"

final_query = f'(with cte as ({original_query}) select *, row_number() over (order by f1) as rn from cte) as q'

Chúng tôi thực hiện hai truy vấn thay vì một:
- truy vấn đầu tiên sẽ được thực hiện để tìm giá trị tối thiểu và tối đa của cột mà chúng tôi sẽ sử dụng làm Cột phân vùng
- trong truy vấn thứ hai, chúng tôi sẽ chuyển kết quả từ truy vấn đầu tiên dưới dạng đối số giới hạn dưới và giới hạn trên .

query_min_max = f"SELECT min(date), max(date) from ({query}) q"
final_query = f"({query}) as q"

df_min_max = spark.read \
    .format("jdbc") \
    .option("driver", "org.postgresql.Driver") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("user", os.environ['user']) \
    .option("password", os.environ['pass']) \
    .option("query", query_min_max) \
    .load()

min = df_min_max.first()["min"]
max = df_min_max.first()["max"]

jdbcDF = spark.read \
    .format("jdbc") \
    .option("driver", "org.postgresql.Driver") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("user", os.environ['user']) \
    .option("password", os.environ['pass']) \
    .option("dbtable", final_query) \
    .option("numPartitions", partitionsCount) \
    .option("partitionColumn", "date") \
    .option("lowerBound", f"{min}") \
    .option("upperBound", f"{max}") \
    .load()

• Nếu bạn trải qua quá trình triển khai Spark trong ứng dụng của mình, hãy cố gắng sử dụng nó ở mức tối đa và đọc dữ liệu đồng thời. Việc thực thi một truy vấn SQL tới spark.read() như được hiển thị trong tài liệu JDBC của Spark sẽ đẩy tất cả dữ liệu vào một phân vùng và chỉ sử dụng một lõi thực thi, bất kể bạn đã thiết lập bao nhiêu lõi trong Phiên Spark của mình.
• Nếu bạn sử dụng Spark ngay hôm nay để đọc dữ liệu từ các nguồn dữ liệu JDBC bằng cách sử dụng phương pháp đơn luồng, hãy thử cấu trúc lại các truy vấn của bạn và cải thiện hiệu suất của chúng, thậm chí bạn có thể được thăng chức khi đang ở đó :)
• Vào cuối ngày, thời gian là tiền bạc. Nếu bạn lo lắng về nỗ lực viết lại các truy vấn của mình, thì tôi có thể nói từ kinh nghiệm — một số truy vấn Spark mà tôi đã tái cấu trúc bằng các kỹ thuật được mô tả ở trên hiện chạy nhanh hơn > 90% so với trước đây. Bạn có thể tưởng tượng sự đón nhận tích cực của cải tiến này đối với nhóm của tôi.

Gabriel