Lời Mở Đầu

Phép JOIN là một trong những lệnh cơ bản và quan trọng nhất khi làm việc với dữ liệu bằng SQL. Trong Apache Spark, cụ thể là SparkSQL, phép JOIN cũng được hỗ trợ như trong các hệ thống cơ sở dữ liệu truyền thống với đầy đủ các loại như INNER JOIN, OUTER JOIN, CROSS JOIN, và nhiều hơn nữa.

Tuy nhiên, bài viết này không chỉ bàn về các loại JOIN mà sẽ tập trung vào các chiến lược thực thi phép JOIN trong Spark. Đưa ra lựa chọn đúng cho chiến lược thực thi là một phần quan trọng trong việc tối ưu hóa hiệu suất chương trình Spark, đặc biệt khi làm việc với các bộ dữ liệu lớn.

Khi thực hiện phép JOIN trong môi trường phân tán như Spark, có một số yếu tố có thể ảnh hưởng đến hiệu suất program:

• Kích thước dữ liệu (Data Size): Khi thực hiện JOIN trên các bộ dữ liệu lớn có thể gây ra hiện tượng Shuffle, đây là một hiện tượng tiêu tốn nguồn lực do phải di chuyển dữ liệu giữa các node trong cluster.
• Dữ liệu lệch (Skew Data): Dữ liệu không được phân bố đều giữa các partition có thể dẫn đến một số node bị quá tải, gây ra các tắc nghẽn hiệu suất.

Các Chiến Lược Thực Thi JOIN Trong Spark

Trong Spark, có một số chiến lược thực thi phép JOIN bao gồm:

• Broadcast Join (Broadcast Hash Join)
• Shuffle Hash Join
• Sort Merge Join
• Skew Join
• Và một vài loại khác ít phổ biến hơn.

Trước khi tìm hiểu về từng loại JOIN, chúng ta cần nắm một số khái niệm quan trọng trong Spark:

• Shuffle: Quá trình tái phân phối dữ liệu giữa các node.
• Hash Table: Cấu trúc dữ liệu sử dụng để thực hiện phép JOIN hiệu quả hơn, và thường được tạo trên dataset nhỏ hơn.

1. Broadcast Join

Broadcast Join thực hiện việc “phát” bảng nhỏ hơn (customer) đến tất cả các node trong cluster, sau đó thực hiện JOIN với bảng lớn hơn (orders) trên mỗi node. Điều này giúp tiết kiệm tài nguyên liên quan đến I/O nhờ giảm thiểu việc di chuyển dữ liệu.

Ví Dụ:

from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast

spark = SparkSession.builder.appName("BroadcastJoinExample").getOrCreate()

customers = [(1, "John Doe"), (2, "Jane Smith"), (3, "Alice Johnson"), (4, "Bob Brown")]
orders = [(101, 1, 250), (102, 2, 450), (103, 1, 150), (104, 3, 300), (105, 4, 200)]

customers_df = spark.createDataFrame(customers, ["CustomerID", "CustomerName"])
orders_df = spark.createDataFrame(orders, ["OrderID", "CustomerID", "Amount"])

joined_df = orders_df.join(broadcast(customers_df), "CustomerID")
joined_df.show()

Lưu Ý:

Độ lớn của dataset để được broadcast có thể được điều chỉnh qua config spark.sql.autoBroadcastJoinThreshold, mặc định là 10MB.

2. Sort Merge Join

Sort Merge Join được thực hiện khi cả hai bảng đều đủ lớn và đã được phân vùng (partitioned) và sắp xếp (sorted) theo khóa JOIN. Spark sẽ shuffle và sort data dựa trên khóa này.

Ví Dụ:

# Initialize Spark session
spark = SparkSession.builder.appName("SortMergeJoinExample").getOrCreate()

customers = [(1, "John Doe"), (2, "Jane Smith"), (3, "Alice Johnson"), (4, "Bob Brown")]
orders = [(101, 1, 250), (102, 2, 450), (103, 1, 150), (104, 3, 300), (105, 4, 200)]

customers_df = spark.createDataFrame(customers, ["CustomerID", "CustomerName"])
orders_df = spark.createDataFrame(orders, ["OrderID", "CustomerID", "Amount"])

joined_df = customers_df.join(orders_df.hint("MERGE"), "CustomerID")
joined_df.show()

3. Shuffle Hash Join

Khi không có bảng nào đủ nhỏ để thực hiện Broadcast, Shuffle Hash Join sẽ được sử dụng. Spark sẽ thực hiện shuffle cả hai bảng và thực hiện Hash JOIN trên mỗi partition.

Ví Dụ:

# Initialize Spark session
spark = SparkSession.builder.appName("ShuffleHashJoinExample").getOrCreate()

customers = [(1, "John Doe"), (2, "Jane Smith"), (3, "Alice Johnson"), (4, "Bob Brown")]
orders = [(101, 1, 250), (102, 2, 450), (103, 1, 150), (104, 3, 300), (105, 4, 200)]

customers_df = spark.createDataFrame(customers, ["CustomerID", "CustomerName"])
orders_df = spark.createDataFrame(orders, ["OrderID", "CustomerID", "Amount"])

joined_df = customers_df.join(orders_df.hint("SHUFFLE_HASH"), "CustomerID")
joined_df.show()

4. Skew Join

Skew Join xảy ra khi dữ liệu được phân bố không đồng đều. Một phương pháp để giải quyết là Salting - thêm giá trị ngẫu nhiên vào cột có dữ liệu skew để cải thiện độ đồng đều trong phân phối.

Ví Dụ:

from pyspark.sql import functions as F
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SkewJoinExample").getOrCreate()

customers = [(1, "John Doe"), (2, "Jane Smith"), (3, "Alice Johnson"), (4, "Bob Brown")]
orders = [(101, 1, 250), (102, 1, 450), (103, 1, 150), (104, 3, 300), (105, 4, 200)]

customers_df = spark.createDataFrame(customers, ["CustomerID", "CustomerName"])
orders_df = spark.createDataFrame(orders, ["OrderID", "CustomerID", "Amount"])

orders_df_salted = orders_df.withColumn("SaltedCustomerID", F.concat(F.col("CustomerID"), F.lit("_"), (F.rand() * 3).cast("int")))
customers_df_salted = customers_df.crossJoin(spark.range(0, 3)).withColumn("SaltedCustomerID", F.concat(F.col("CustomerID"), F.lit("_"), F.col("id"))).drop("id")
joined_df = orders_df_salted.join(customers_df_salted, "SaltedCustomerID").drop("SaltedCustomerID")
joint_df.show()

Kết Luận

Lựa chọn chiến lược thực thi phù hợp khi thực hiện phép JOIN trong Apache Spark là rất cần thiết để tối ưu hóa hiệu suất và hiệu quả công việc. Hy vọng bài viết này sẽ giúp bạn hiểu rõ hơn về các chiến lược JOIN trong Spark và áp dụng chúng một cách hiệu quả trong các dự án của mình.

Tham Khảo

• Kỹ thuật tối ưu hóa Spark: Các loại JOIN
• Chiến lược JOIN trong Apache Spark — Cách tiếp cận thực tiễn!
  source: viblo