PySpark DataFrameの行ごとの集約

Question

行ごとの操作を行う関数を使用して集計するPyspark DataFrameがあります。

私は4列を有し、列AI内の各一意の値の列B、C、D

に行ごと凝集を行う必要があり、私はこの方法を使用しています：

1. を

   は使用して一意の値を取得

   A_uniques = df.select('A').distinct() 
   

2. def func(x): 
       y = df.filter(df.A==x) 
       y = np.array(y.toPandas()) 
       for i in y.shape[0]: 
        y[i,1] = y[i-1,0] 
        y[i,0] = (y[i,0]+y[i,2])/y[i,3] 
       agg = sum(y[:,1]) 
       return agg 
   

3. A_uniques.rdd.map(lambda x: (x['A'], func(x['A']))) 
   

私はこのエラーを取得しています：

PicklingError: Could not serialize object: Py4JError: An error occurred while calling o64.getnewargs. Trace: py4j.Py4JException: Method getnewargs([]) does not exist at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318) at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:326) at py4j.Gateway.invoke(Gateway.java:272) at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at py4j.commands.CallCommand.execute(CallCommand.java:79) at py4j.GatewayConnection.run(GatewayConnection.java:214) at java.lang.Thread.run(Thread.java:748)

はRDDSでnumpyの配列を保存する解決策はありますか？または、この操作全体を他の方法で行うことはできますか？

Answer 1

Pysparkでは、groupBy（）（私の場合は2 colsでグループ化）関数を使用してGroupedDataFrameを取得し、agg()関数をパイプします。私はリストに集約されたデータを変換し、それらと協力しin this example、あなたはまたGroupedDataFrameためgrouped_Series_Owner = x_gb["Owner"].apply(list) .apply（）関数を使用することができます...以下

sqlContext.sql("select * from retail_db.orders").groupBy("order_status", "order_date").agg({"order_customer_id": "sum", "order_id": "count"}).show() 

+---------------+--------------------+----------------------+---------------+ 
| order_status|   order_date|sum(order_customer_id)|count(order_id)| 
+---------------+--------------------+----------------------+---------------+ 
|PENDING_PAYMENT|2013-07-28 00:00:...|    237876|    37| 
|  COMPLETE|2013-08-22 00:00:...|    415843|    64| 
|PENDING_PAYMENT|2013-10-20 00:00:...|    168223|    28| 
|SUSPECTED_FRAUD|2013-11-22 00:00:...|     36354|    6| 
|PENDING_PAYMENT|2013-12-19 00:00:...|    131972|    22| 
|PENDING_PAYMENT|2014-03-12 00:00:...|    352832|    52| 
|  ON_HOLD|2014-03-28 00:00:...|     74970|    13| 
|SUSPECTED_FRAUD|2014-04-14 00:00:...|     18145|    2| 
|  PENDING|2014-04-21 00:00:...|    174419|    26| 
|   CLOSED|2014-06-04 00:00:...|     66677|    10| 
|PENDING_PAYMENT|2014-06-26 00:00:...|    249542|    45| 
|PENDING_PAYMENT|2013-08-17 00:00:...|    405980|    56| 
|   CLOSED|2013-09-10 00:00:...|    164670|    23| 
|SUSPECTED_FRAUD|2013-09-19 00:00:...|     26613|    4| 
|  PENDING|2013-09-26 00:00:...|    176547|    28| 
|  COMPLETE|2013-10-20 00:00:...|    314462|    54| 
|  CANCELED|2013-10-31 00:00:...|     36881|    6| 
|  PROCESSING|2013-11-09 00:00:...|    149164|    23| 
| PAYMENT_REVIEW|2013-11-29 00:00:...|     17368|    3| 
|SUSPECTED_FRAUD|2013-12-11 00:00:...|     45085|    7| 
+---------------+--------------------+----------------------+---------------+ 
only showing top 20 rows 

を例を参照してください。