2. 程式人生 > >pyspark 自定義聚合函式 UDAF

pyspark 自定義聚合函式 UDAF

阿新 • • 發佈：2018-12-21

自定義聚合函式 UDAF 目前有點麻煩，PandasUDFType.GROUPED_AGG 在2.3.2的版本中不知怎麼回事，不能使用！

這樣的話只能曲線救國了！

 

 

PySpark有一組很好的聚合函式（例如，count，countDistinct，min，max，avg，sum），但這些並不適用於所有情況（特別是如果你試圖避免代價高昂的Shuffle操作）。

PySpark目前有pandas_udfs，它可以建立自定義聚合器，但是你一次只能“應用”一個pandas_udf。如果你想使用多個，你必須預先形成多個groupBys ......並且避免那些改組。

在這篇文章中，我描述了一個小黑客，它使您能夠建立簡單的python UDF，它們對聚合資料起作用（此功能只應存在於Scala中！）。

1 
2 
3 
4 
5 
6 7 8 
from pyspark.sql import functions as F from pyspark.sql import types as T  a = sc.parallelize([[1, 'a'],  [1, 'b'],  [1, 'b'],  [2, 'c']]).toDF(['id', 'value']) a.show()
ID 值
1 '一個'
1
 'B'
1 'B'
2 'C'

我使用collect_list將給定組中的所有資料放入一行。我列印下面這個操作的輸出。

1

a.groupBy('id').agg(F.collect_list('value').alias('value_list')).show()
ID VALUE_LIST
1 ['a'，'b'，'b']
2 ['C']

然後我建立一個UDF，它將計算這些列表中字母'a'的所有出現（這可以很容易地在沒有UDF的情況下完成，但是你明白了）。此UDF包含collect_list，因此它作用於collect_list的輸出。

1 
2 
3 
4 
5 
6 7 8 9 10 11 
def find_a(x):  """Count 'a's in list."""  output_count = 0  for i in x:  if i == 'a':  output_count += 1  return output_count  find_a_udf = F.udf(find_a, T.IntegerType())  a.groupBy('id').agg(find_a_udf(F.collect_list('value')).alias('a_count')).show()
ID A_COUNT
1 1
2 0

我們去！作用於聚合資料的UDF！接下來，我展示了這種方法的強大功能，結合何時讓我們控制哪些資料進入F.collect_list。

首先，讓我們建立一個帶有額外列的資料框。

1 
2 
3 
4 
5 
6 7 8 9 
from pyspark.sql import functions as F from pyspark.sql import types as T  a = sc.parallelize([[1, 1, 'a'],  [1, 2, 'a'],  [1, 1, 'b'],  [1, 2, 'b'],  [2, 1, 'c']]).toDF(['id', 'value1', 'value2']) a.show()
ID 值1 值2
1 1 '一個'
1 2 '一個'
1 1 'B'
1 2 'B'
2 1 'C'

請注意，我如何在collect_list中包含一個when。請注意，UDF仍然包含collect_list。

1

a.groupBy('id').agg(find_a_udf( F.collect_list(F.when(F.col('value1') == 1, F.col('value2')))).alias('a_count')).show()
ID A_COUNT
1 1
2 0

https://danvatterott.com/blog/2018/09/06/python-aggregate-udfs-in-pyspark/

 

 

 

還有一種做法就是用pandas_udf， 然後按照聚合的得到的列去重

相關推薦

pyspark 自定義聚合函式 UDAF

自定義聚合函式 UDAF 目前有點麻煩，PandasUDFType.GROUPED_AGG 在2.3.2的版本中不知怎麼回事，不能使用！ 這樣的話只能曲線救國了！     PySpark有一組很好的聚合函式（例如，count，countDistinct，min，max，avg，su

大資料學習之路90-sparkSQL自定義聚合函式UDAF

什麼是UDAF？就是輸入N行得到一個結果，屬於聚合類的。 接下來我們就寫一個求幾何平均數的一個自定義聚合函式的例子 我們從開頭寫起，先來看看需要進行計算的數如何產生： package com.test.SparkSQL import java.lang import org

spark1.5 自定義聚合函式UDAF

自定義聚合函式需要實現UserDefinedAggregateFunction，以下是該抽象類的定義，加了一點註釋： /* * Licensed to the Apache Software Foundation (ASF) under one or more * co

Hive使用者自定義聚合函式(UDAF)

使用者自定義聚合函式，使用者提供的多個入參通過聚合計算(求和、求最大值、求最小值)得到一個聚合計算結果的函式。 問題：UDF也可以提供輸入多個引數然後輸出一個結果的運算，比如加法運算add(3，5)，add這個UDF需要實現UDF的evaluate方法,那麼UDF和UD

14.Spark SQL：UDAF自定義聚合函式實戰

UDAF自定義函式實戰 UDAF：User Defined Aggregate Function。使用者自定義聚合函式。是Spark 1.5.x引入的最新特性。 UDF，其實更多的是針對單行輸入，返

SparkSQL建立RDD：UDAF（UserDefinedAggregatedFunction）使用者自定義聚合函式【Java版純程式碼】

要實現8個方法，8個方法中，最為重要的有3個： initialize：初始化，在給，map端每一個分割槽的每一個key進行初始化，給0 update：在map端聚合 merge： 在reduce端聚合 Java版程式碼： package com.bjsxt; im

Hive通用型自定義聚合函式（UDAF）

計算器必須實現的方法： 1、getNewAggregationBuffer()：返回儲存臨時聚合結果的AggregationBuffer物件。 2、reset(AggregationBuffer agg)：重置聚合結果物件，以支援mapper和reducer的重用。 3、iterate(Aggregati

Oracle 自定義聚合函式

create or replace type str_concat_type as object ( cat_string varchar2(4000), static function ODCIAggregateInitialize(cs_ctx In Out str_concat_

PostgreSql的聚合函式string_agg和自定義聚合函式group_concat

1.使用自定義聚合函式  group_concat SELECT * FROM ( SELECT "FFID",CAST ("BRDNO" AS INTEGER),"count" (*) AS "NUM", array_to_string(g

postgresql 9.3 自定義聚合函式實現多行資料合併成一列

前言 常見的一種需求，如下圖(1)： 目前需要按右邊的factor_code進行分組，將左邊的behavior_code這一列通過指定分隔符連線起來，比如通過<br /> 來連線，理想的效果應當是如下圖(2)這樣： 下面就來討論一下實現

ORACLE中的自定義聚合函式（User-Defined Aggregate Functions）

聚合函式也叫分組函式，是對多行進行計算的一種函式，比如SUM、COUNT、AVG、MAX、MIN等等。Oracle中為了增強擴充套件性，提供了一個自定義聚合函式的介面，ODCIAggregate()，使用者可以用Oracle支援的各類程式語言（PLSQL、C、C++、JAVA

tensflow自定義損失函式

三、自定義損失函式 標準的損失函式並不合適所有場景，有些實際的背景需要採用自己構造的損失函式，Tensorflow 也提供了豐富的基礎函式供自行構建。 例如下面的例子：當預測值（y_pred）比真實值（y_true）大時，使用 (y_pred-y_true)*loss_more 作為 loss，

16 友盟專案--日期工具類、UTF自定義日期函式

日期工具類、UTF自定義日期函式       日期函式工具類 DateUtil 以date為參照，查詢指定按照day、Month、week進行偏移日期時間格式

自定義sort函式排序標準

function compareStandard(key){ return function(obj1,obj2){ let val1=obj1[key]; let val2=obj2[key]; if (val1<val2) {

自定義Jquery函式和函式回撥

一：使用Jquery自定義函式 1、引入Jquery外掛    <script type="text/javascript" src="jquery-2.1.0.min.js"></script>   2、定義格式 開頭：(f

pytorch系列12 --pytorch自定義損失函式custom loss function

本文主要內容： nn.Module 和 nn.Functional 區別和聯絡 自定義損失函式 1. 關於nn.Module與nn.Functional的區別： https://discuss.pytorch.org/t/whats-the-differe

c++ set自定義比較函式

set<char*>無法像set<string>一樣進行預設的排序操作，需要自己定義比較函式進行排序，例子如下： #include <stdio.h> #include "stdafx.h" #include<iostream> #incl

如何自定義sort函式中的比較函式

參考連結 題目描述 輸入一個正整數陣列，把數組裡所有數字拼接起來排成一個數，列印能拼接出的所有數字中最小的一個。例如輸入陣列{3，32，321}，則打印出這三個數字能排成的最小數字為321323。 思路：自定義比較器，若a+b>b+a則a>b，即”3”+”23”>

Oracle 自定義聚合函數

end acl ember ins .com iter icons rate nbsp create or replace type str_concat_type as object ( cat_string varchar2(4000), sta

C++11使用自定義hash函式及比較函式的unordered_set

#include <unordered_set> #include <functional> #include <iostream> struct MyKey { int key; }; struct MyKeyHashHasher { size_