使用PySpark處理Hive中的每個組的數據

Question

我有一些用戶數據存儲在Hive表（ORC文件格式）中，每個用戶有多行。例如：

user | attr1 | attr2 
---- | ----- | ----- 
u1 | 1  | a 
u1 | 2  | b 
u2 | 3  | a 
u2 | 4  | b 

我需要映射屬於同一用戶的每一組行，以計算一些字符串。在這個例子中，f([(1, a), (2, b)])和f([(3, a), (4, b)])。

這怎麼可以使用Hive + Spark的Python API完成？

從我能看到的Python GroupedData API只有stat函數。我應該採取什麼措施來減少潛在的行RDD？有更有效的方式使用Hive/ORC柱狀格式進行分組嗎？

Answer 1

from pyspark.sql import functions as f 

df = spark.createDataFrame([['u1', 1, 'a'], ['u1', 2, 'b'], ['u2', 3, 'a'], ['u2', 4, 'b']], 
          schema=['user', 'attr1', 'attr2']) 
df = df.groupBy('user').agg(f.collect_list(f.array(df['attr1'], df['attr2'])).alias('array_col')) 
df = df.withColumn('len_col', f.size(df['new_col'])) 
df.show(truncate=False) 

+----+----------------------------------------+-------+       
|user|array_col        |len_col| 
+----+----------------------------------------+-------+ 
|u1 |[WrappedArray(1, a), WrappedArray(2, b)]|2  | 
|u2 |[WrappedArray(3, a), WrappedArray(4, b)]|2  | 
+----+----------------------------------------+-------+