Nhiệm vụ cuối cùng của Spark mất 100x lần dài hơn 199 đầu tiên, cách cải thiện

Question

Tôi thấy một số vấn đề về hiệu suất trong khi chạy truy vấn bằng cách sử dụng các khung dữ liệu. Tôi đã thấy trong nghiên cứu của tôi, rằng nhiệm vụ cuối cùng chạy lâu có thể là dấu hiệu cho thấy dữ liệu không bị quấy rầy một cách tối ưu, nhưng chưa tìm thấy một quy trình chi tiết để giải quyết vấn đề này.

Tôi bắt đầu tải hai bảng dưới dạng các khung dữ liệu và sau đó tôi sẽ tham gia các bảng đó trên một trường. Tôi đã cố gắng để thêm phân phối bởi (phân vùng), và sắp xếp theo, để cải thiện hiệu suất, nhưng tôi vẫn thấy nhiệm vụ cuối cùng dài chạy duy nhất này. Đây là một phiên bản đơn giản của mã của tôi, lưu ý rằng một và hai truy vấn không thực sự đơn giản và sử dụng các UDF này để tính toán một số giá trị.

Tôi đã thử một vài cài đặt khác nhau cho spark.sql.shuffle. Tôi đã thử 100, nhưng nó thất bại (tôi đã không thực sự gỡ lỗi này đến nhiều để được trung thực). Tôi đã thử 300, 4000 và 8000. Hiệu suất giảm với mỗi lần tăng. Tôi đang chọn một ngày dữ liệu, trong đó mỗi tệp là một giờ.

val df1 = sqlContext.sql("Select * from Table1") 
val df2 = sqlContext.sql("Select * from Table2") 

val distributeDf1 = df1 
    .repartition(df1("userId")) 
    .sortWithinPartitions(df1("userId")) 

val distributeDf2 = df2 
    .repartition(df2("userId")) 
    .sortWithinPartitions(df2("userId")) 

distributeDf1.registerTempTable("df1") 
distributeDf2.registerTempTable("df2") 

val df3 = sqlContext 
    .sql(""" 
    Select 
     df1.* 
    from 
     df1 
    left outer join df2 on 
     df1.userId = df2.userId""") 

Do dường như phân vùng theo userId không lý tưởng nên tôi có thể phân vùng bằng dấu thời gian thay thế. Nếu tôi làm điều này, tôi có nên làm Date + Hour không? Nếu tôi có ít hơn 200 combo độc đáo cho điều này, liệu tôi sẽ có những người thực thi trống không?

Answer 1

Bạn rõ ràng gặp sự cố với số liệu lớn nghiêng. Cho phép hãy xem một sự statistics you've provided:

df1 = [mean=4.989209978967438, stddev=2255.654165352454, count=2400088] 
df2 = [mean=1.0, stddev=0.0, count=18408194] 

Với trung bình khoảng 5 và độ lệch chuẩn hơn 2000 bạn nhận được một đuôi dài.

Vì một số phím thường xuyên hơn nhiều so với các phím khác sau khi phân vùng lại một số người thực thi sẽ có nhiều việc phải làm hơn so với các phím còn lại.

Hơn nữab mô tả của bạn cho thấy rằng sự cố có thể xảy ra với một hoặc một vài phím băm đến cùng một phân vùng.

Vì vậy, hãy đầu tiên của xác định giá trị ngoại biên (giả):

val mean = 4.989209978967438 
val sd = 2255.654165352454 

val df1 = sqlContext.sql("Select * from Table1") 
val counts = df.groupBy("userId").count.cache 

val frequent = counts 
    .where($"count" > mean + 2 * sd) // Adjust threshold based on actual dist. 
    .alias("frequent") 
    .join(df1, Seq("userId")) 

và phần còn lại:

val infrequent = counts 
    .where($"count" <= mean + 2 * sd) 
    .alias("infrequent") 
    .join(df1, Seq("userId")) 

Là nó thực sự một cái gì đó để được mong đợi? Nếu không, hãy cố gắng xác định nguồn gốc của vấn đề ở phía trên.

Nếu nó được mong đợi, bạn có thể thử:

• phát sóng bảng nhỏ:

  val df2 = sqlContext.sql("Select * from Table2") 
  df2.join(broadcast(df1), Seq("userId"), "rightouter") 
  

• tách, hợp nhất (union) và phát sóng chỉ thường xuyên:

  df2.join(broadcast(frequent), Seq("userId"), "rightouter") 
      .union(df2.join(infrequent, Seq("userId"), "rightouter")) 
  

• muối userId với một số dữ liệu ngẫu nhiên

nhưng bạn không nên:

• phân vùng lại toàn bộ dữ liệu và sắp xếp tại địa phương (mặc dù sắp xếp tại địa phương mình không phải là một vấn đề)
• thực hiện hash tiêu chuẩn tham gia vào dữ liệu đầy đủ.