Spark DataFrame: đếm các giá trị riêng biệt của mỗi cột

Question

Câu hỏi này có khá nhiều trong tiêu đề: Có cách hiệu quả để tính các giá trị riêng biệt trong mỗi cột trong một DataFrame không?

Phương thức describe chỉ cung cấp số đếm nhưng không tính số khác biệt và tôi tự hỏi liệu có cách nào để nhận số đếm riêng biệt cho tất cả các cột (hoặc một số được chọn) hay không.

Answer 1

Nhiều quy tụ sẽ là khá đắt tiền để tính toán, vì vậy tôi muốn khuyên bạn nên sử dụng xấp xỉ đếm riêng biệt:

val df = Seq((1,3,4),(1,2,3),(2,3,4),(2,3,5)).toDF("col1","col2","col3") 

val exprs = df.columns.map((_ -> "approx_count_distinct")).toMap 
df.agg(exprs).show() 
// +---------------------------+---------------------------+---------------------------+ 
// |approx_count_distinct(col1)|approx_count_distinct(col2)|approx_count_distinct(col3)| 
// +---------------------------+---------------------------+---------------------------+ 
// |       2|       2|       3| 
// +---------------------------+---------------------------+---------------------------+ 

Phương pháp approx_count_distinct dựa vào HyperLogLog dưới mui xe.

Thuật toán HyperLogLog và biến thể của nó HyperLogLog ++ (được thực hiện trong Spark) dựa trên quan sát thông minh sau đây.

Nếu các con số được trải đều trên một phạm vi, thì số lượng các phần tử riêng biệt có thể xấp xỉ từ số 0 đứng đầu lớn nhất trong biểu diễn nhị phân của các số.

Ví dụ: nếu chúng ta quan sát một số có chữ số ở dạng nhị phân có dạng 0…(k times)…01…1, thì chúng tôi có thể ước tính rằng có thứ tự của 2^k phần tử trong tập hợp. Đây là một ước tính rất thô sơ nhưng nó có thể được tinh chế với độ chính xác cao với một thuật toán phác thảo.

Giải thích kỹ lưỡng về các cơ chế đằng sau thuật toán này có thể được tìm thấy trong original paper.

Lưu ý: khởi điểm Spark 1.6, khi Spark gọi SELECT SOME_AGG(DISTINCT foo)), SOME_AGG(DISTINCT bar)) FROM df mỗi khoản nên kích hoạt tập hợp riêng cho từng khoản. Trong khi điều này khác với SELECT SOME_AGG(foo), SOME_AGG(bar) FROM df nơi chúng tôi tổng hợp một lần. Do đó hiệu suất sẽ không thể so sánh được khi sử dụng count(distinct(_)) và approxCountDistinct (hoặc approx_count_distinct).

Đó là một trong những thay đổi về hành vi kể từ Spark 1.6:

Với kế hoạch truy vấn cải thiện cho các truy vấn có aggregations biệt (SPARK-9241), kế hoạch của một truy vấn có một tập hợp riêng biệt duy nhất có đã được thay đổi thành một phiên bản mạnh mẽ hơn. Để chuyển về kế hoạch được tạo bởi kế hoạch của Spark 1.5, vui lòng đặt spark.sql.specializeSingleDistinctAggPlanning thành true. (SPARK-12077)

Tham chiếu: Approximate Algorithms in Apache Spark: HyperLogLog and Quantiles.

Answer 2

Trong pySpark bạn có thể làm một cái gì đó như thế này, sử dụng countDistinct():

from pyspark.sql.functions import col, countDistinct 

df.agg(*(countDistinct(col(c)).alias(c) for c in df.columns)) 

Tương tự như vậy trong Scala:

import org.apache.spark.sql.functions.countDistinct 
import org.apache.spark.sql.functions.col 

df.select(df.columns.map(c => countDistinct(col(c)).alias(c)): _*) 

Nếu bạn muốn điều tốc độ lên ở khả năng mất tính chính xác, bạn có thể cũng sử dụng approxCountDistinct().

Answer 3

nếu bạn chỉ muốn đếm cột perticular thì sau đây có thể hữu ích. Mặc dù câu trả lời muộn của nó. nó có thể giúp ai đó. (pyspark 2.2.0 được kiểm tra)

from pyspark.sql.functions import col, countDistinct 
df.agg(countDistinct(col("colName")).alias("count")).show()