Chức năng Spark so với hiệu năng UDF?

Question

Spark hiện cung cấp các hàm được xác định trước có thể được sử dụng trong các khung dữ liệu và dường như chúng được tối ưu hóa cao. Câu hỏi ban đầu của tôi sẽ diễn ra nhanh hơn, nhưng tôi đã tự mình thử nghiệm và tìm thấy các hàm tia lửa nhanh hơn khoảng 10 lần trong một trường hợp. Có ai biết tại sao điều này là như vậy, và khi nào một udf sẽ nhanh hơn (chỉ cho các trường hợp mà một chức năng spark giống hệt nhau tồn tại)?

Đây là mã thử nghiệm của tôi (chạy trên Databricks ed cộng đồng): function

# UDF vs Spark function 
from faker import Factory 
from pyspark.sql.functions import lit, concat 
fake = Factory.create() 
fake.seed(4321) 

# Each entry consists of last_name, first_name, ssn, job, and age (at least 1) 
from pyspark.sql import Row 
def fake_entry(): 
    name = fake.name().split() 
    return (name[1], name[0], fake.ssn(), fake.job(), abs(2016 - fake.date_time().year) + 1) 

# Create a helper function to call a function repeatedly 
def repeat(times, func, *args, **kwargs): 
    for _ in xrange(times): 
     yield func(*args, **kwargs) 
data = list(repeat(500000, fake_entry)) 
print len(data) 
data[0] 

dataDF = sqlContext.createDataFrame(data, ('last_name', 'first_name', 'ssn', 'occupation', 'age')) 
dataDF.cache() 

UDF:

concat_s = udf(lambda s: s+ 's') 
udfData = dataDF.select(concat_s(dataDF.first_name).alias('name')) 
udfData.count() 

Spark Chức năng:

spfData = dataDF.select(concat(dataDF.first_name, lit('s')).alias('name')) 
spfData.count() 

Ran cả nhiều lần, udf thường mất khoảng 1,1 - 1,4 s và chức năng spark concat luôn dưới 0,15 giây.

Answer 1

khi một udf sẽ nhanh

Nếu bạn hỏi về Python UDF câu trả lời có lẽ là không bao giờ. Vì các hàm SQL tương đối đơn giản và không được thiết kế cho các nhiệm vụ phức tạp nên không thể bù đắp được chi phí lặp đi lặp lại, deserialization và chuyển động dữ liệu giữa trình thông dịch Python và JVM.

Có ai biết tại sao điều này là rất

Những lý do chính đã được liệt kê ở trên và có thể được giảm đến một thực tế đơn giản rằng Spark DataFrame là natively một cấu trúc JVM và phương pháp truy cập tiêu chuẩn được thực hiện bởi đơn giản các cuộc gọi đến Java API. UDF mặt khác được thực hiện bằng Python và yêu cầu di chuyển dữ liệu qua lại.

Trong khi PySpark nói chung yêu cầu chuyển động dữ liệu giữa JVM và Python, trong trường hợp API RDD mức thấp, nó thường không yêu cầu hoạt động serde đắt tiền. Spark SQL bổ sung thêm chi phí serialization và serialization cũng như chi phí di chuyển dữ liệu từ và đến đại diện không an toàn trên JVM. Cái sau là đặc trưng cho tất cả các UDF (Python, Scala và Java) nhưng cái trước đây là đặc trưng cho các ngôn ngữ không phải là ngôn ngữ bản địa.

Không giống như UDF, chức năng Spark SQL hoạt động trực tiếp trên JVM và thường được tích hợp tốt với cả Catalyst và Tungsten. Nó có nghĩa là chúng có thể được tối ưu hóa trong kế hoạch thực hiện và hầu hết thời gian có thể được hưởng lợi từ codgen và các tối ưu hóa Tungsten khác. Hơn nữa, chúng có thể hoạt động trên dữ liệu trong biểu diễn "tự nhiên" của nó.

Vì vậy, theo một nghĩa nào đó, vấn đề ở đây là Python UDF phải mang dữ liệu đến mã trong khi các biểu thức SQL đi theo cách khác.

Answer 2

Kể từ ngày 30 tháng 10 năm 2017, Spark vừa giới thiệu các tệp pdf được vector hóa cho pyspark.

https://databricks.com/blog/2017/10/30/introducing-vectorized-udfs-for-pyspark.html

Lý do mà Python UDF là chậm, có lẽ là PySpark UDF không được thực hiện một cách tối ưu nhất:

Theo đoạn từ liên kết.

Spark thêm một API Python trong phiên bản 0.7, với hỗ trợ cho các hàm do người dùng xác định. Các hàm do người dùng định nghĩa này hoạt động một hàng-at-a-time và do đó phải chịu sự tuần tự hóa cao và phí trên cao.

Tuy nhiên các file PDF mới vector hóa dường như được cải thiện hiệu suất rất nhiều:

từ 3x đến hơn 100x.