{"id":372471,"date":"2024-05-21T05:13:00","date_gmt":"2024-05-21T05:13:00","guid":{"rendered":"http:\/\/savepearlharbor.com\/?p=372471"},"modified":"-0001-11-30T00:00:00","modified_gmt":"2024-05-21T05:13:00","slug":"","status":"publish","type":"post","link":"https:\/\/savepearlharbor.com\/?p=372471","title":{"rendered":"<span>User-defined aggregation functions in Spark<\/span>"},"content":{"rendered":"<div><!--[--><!--]--><\/div>\n<div id=\"post-content-body\">\n<div>\n<div class=\"article-formatted-body article-formatted-body article-formatted-body_version-2\">\n<div xmlns=\"http:\/\/www.w3.org\/1999\/xhtml\">\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/fd3\/6c7\/e5d\/fd36c7e5ddad88b03d593266ed3dea4d.png\" width=\"1432\" height=\"806\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/fd3\/6c7\/e5d\/fd36c7e5ddad88b03d593266ed3dea4d.png\"\/><\/figure>\n<p>Below, we will discuss user-defined aggregation functions (UDAF) using <code>org.apache.spark.sql.expressions.Aggregator<\/code><strong>,<\/strong> which can be used for aggregating groups of elements in a DataSet into a single value in any user-defined way.<\/p>\n<p>Let\u2019s start by examining an example from the official documentation that implements a simple aggregation<\/p>\n<pre><code class=\"scala\">case class Data(i: Int)  val customSummer =  new Aggregator[Data, Int, Int] {      def zero: Int = 0      def reduce(b: Int, a: Data): Int = b + a.i      def merge(b1: Int, b2: Int): Int = b1 + b2      def finish(r: Int): Int = r      def bufferEncoder: Encoder[Int] = Encoders.scalaInt      def outputEncoder: Encoder[Int] = Encoders.scalaInt    }.toColumn()  val ds: Dataset[Data] = ... val aggregated = ds.select(customSummer)<\/code><\/pre>\n<p>The case class \u2018Data\u2019 is needed because UDAFs are used for DataSets.<\/p>\n<p>To create your own aggregator, you need to define 6 functions with predefined names:<\/p>\n<ul>\n<li>\n<p><strong>zero<\/strong> \u2014 also known as the initial value; should satisfy the requirement: \u201csomething\u201d + zero = \u201csomething\u201d<\/p>\n<\/li>\n<li>\n<p><strong>reduce<\/strong> \u2014 the reduce function that performs our aggregation into a buffer<\/p>\n<\/li>\n<li>\n<p><strong>merge<\/strong> \u2014 the function for merging buffers<\/p>\n<\/li>\n<li>\n<p><strong>finish<\/strong>  \u2014 the final processing to obtain the target value; in the specific  example, it is essentially absent, but it is often useful (as we\u2019ll see  in the examples below)<\/p>\n<\/li>\n<li>\n<p>two <strong>encoders<\/strong> for the buffer and the output value<\/p>\n<\/li>\n<\/ul>\n<p>Why  so many? Let\u2019s dive a little bit deeper. Spark is a distributed data  processing framework and processes aggregation in the following steps:<\/p>\n<ul>\n<li>\n<p><strong>First step:<\/strong> On the executors, we perform a pre-aggregation using a reduce function to minimize the volume of data shuffling.<\/p>\n<\/li>\n<li>\n<p><strong>Second step:<\/strong> Spark shuffles the data \u2014 moving the data with the same aggregation key to the same executor.<\/p>\n<\/li>\n<li>\n<p><strong>Third step: <\/strong>Finalizes aggregation on executors using a merge function.<\/p>\n<\/li>\n<li>\n<p><strong>Fourth step:<\/strong> Final processing using a finish function.<\/p>\n<\/li>\n<\/ul>\n<p>Therefore,  even if you don\u2019t need a certain step, we still have to define these  functions. On the other hand, you are unlikely to use UDAF for simple  summation, and for complex aggregations, such staging can be very  useful. Also, it is important to note that the aggregation is  essentially performed twice \u2014 first on the executors, and then the final  aggregation is done with data shuffling.<\/p>\n<p>Let\u2019s  consider an educational \u2018word count\u2019 example implemented using  aggregators. Suppose we have a simple CSV file with user tweets:<\/p>\n<pre><code>userId,tweet f6e8252f,cat dog frog cat dog frog frog dog dog bird f6e8252f,cat dog frog dog dog bird f6e8252f,cat dog 29d89ee4,frog frog dog bird 29d89ee4,frog cat dog frog frog dog dog bird 29d89ee4,frog bird<\/code><\/pre>\n<p>Let\u2019s  determine the most frequently used word for each user. Naturally, there  are many ways to solve this task, but for educational purposes, we will  demonstrate how to solve it using a UDAF.<\/p>\n<pre><code class=\"scala\">val df = spark.read.format(\"csv\").option(\"header\", \"true\").load(path)  type FrenquencyMapType = Map[String, Int]  class MyCustomAggregator(columnName: String) extends Aggregator [Row, FrenquencyMapType, String] {    private def combineMaps(map1: FrenquencyMapType, map2: FrenquencyMapType): FrenquencyMapType = {     val combinedKeys = map1.keySet ++ map2.keySet     combinedKeys.map { key => key -> (map1.getOrElse(key, 0) + map2.getOrElse(key, 0))     }.toMap   }    def zero: FrenquencyMapType = Map.empty[String, Int]    def reduce(buffer: FrenquencyMapType, row: Row): FrenquencyMapType = {     val wordsMap = row.getAs[Any](columnName).toString.split(\" \").groupBy(identity).mapValues(_.length)     combineMaps(buffer, wordsMap)    }    def merge(map1: FrenquencyMapType, map2: FrenquencyMapType): FrenquencyMapType =     combineMaps(map1, map2)    def finish(buffer: FrenquencyMapType): String =     if (buffer.isEmpty) \"nUll\" else buffer.maxBy(_._2)._1   def bufferEncoder: Encoder[FrenquencyMapType] = Encoders.kryo[Map[String, Int]]    def outputEncoder: Encoder[String] = Encoders.STRING  }  val mostPopularWordAggregator = new MyCustomAggregator(\"tweet\").toColumn  df.groupBy(col(\"userID\"))   .agg(mostPopularWordAggregator.name(\"favoriteWord\"))   .withColumnRenamed(\"value\", \"userId\")   .show()<\/code><\/pre>\n<p>The result:<\/p>\n<pre><code>+--------+---------------+ |userId  |favoriteWord   | +--------+---------------+ |29d89ee4|frog           | |f6e8252f|dog            | +--------+---------------+<\/code><\/pre>\n<p>Note the specification of types &lt;IN, BUF, OUT>:<\/p>\n<ul>\n<li>\n<p>The input type for our aggregator is Row.<\/p>\n<\/li>\n<li>\n<p>For the buffer, we use the type <code>Map[String, Int]<\/code>(alias <strong>FrenquencyMapType<\/strong>)<\/p>\n<\/li>\n<li>\n<p>Finally, the output type is simply String since the most popular word is returned.<\/p>\n<\/li>\n<\/ul>\n<p>Astute  readers may have noticed that, in essence, we adapted the \u2018most common  value\u2019 aggregator for our needs. Could we achieve the same result  without UDAF? Of course! One of the naives variants is:<\/p>\n<ol>\n<li>\n<p>Get an ArrayType column using <code>org.apache.spark.sql.functions.split<\/code><\/p>\n<\/li>\n<li>\n<p>Create a new row for each element in the array using <code>org.apache.spark.sql.functions.explode<\/code><\/p>\n<\/li>\n<li>\n<p>Perform standard aggregation over <strong>userId + word<\/strong>.<\/p>\n<\/li>\n<li>\n<p>Take the <strong>userId + word<\/strong> with max value of count-aggregation.<\/p>\n<\/li>\n<\/ol>\n<p>In the code it could look like this:<\/p>\n<pre><code class=\"scala\">val window = Window.partitionBy(\"userId\")  df.withColumn(\"arrayOfWords\", split(col(\"tweet\"), \" \"))   .withColumn(\"word\", explode(col(\"arrayOfWords\")))   .groupBy(\"userId\", \"word\")   .agg(count(\"word\").as(\"count\"))   .withColumn(\"maxCount\", max(\"count\").over(window))   .filter(col(\"count\") === col(\"maxCount\"))   .select(\"userId\", \"word\")   .show()<\/code><\/pre>\n<p>And now let\u2019s compare the plans. With UDAF it\u2019s very simple:<\/p>\n<pre><code>== Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- ObjectHashAggregate(keys=[userID#16], functions=[mycustomaggregator($line15.$read$$iw$$iw$$iw$$iw$MyCustomAggregator@69cf253c, Some(createexternalrow(userId#16.toString, tweet#17.toString, StructField(userId,StringType,true), StructField(tweet,StringType,true))), Some(interface org.apache.spark.sql.Row), Some(StructType(StructField(userId,StringType,true), StructField(tweet,StringType,true))), encodeusingserializer(input[0, java.lang.Object, true], true), decodeusingserializer(input[0, binary, true], scala.collection.immutable.Map, true), staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true, false), StringType, true, 0, 0)])    +- Exchange hashpartitioning(userID#16, 200), ENSURE_REQUIREMENTS, [id=#30]       +- ObjectHashAggregate(keys=[userID#16], functions=[partial_mycustomaggregator($line15.$read$$iw$$iw$$iw$$iw$MyCustomAggregator@69cf253c, Some(createexternalrow(userId#16.toString, tweet#17.toString, StructField(userId,StringType,true), StructField(tweet,StringType,true))), Some(interface org.apache.spark.sql.Row), Some(StructType(StructField(userId,StringType,true), StructField(tweet,StringType,true))), encodeusingserializer(input[0, java.lang.Object, true], true), decodeusingserializer(input[0, binary, true], scala.collection.immutable.Map, true), staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true, false), StringType, true, 0, 0)])          +- FileScan csv [userId#16,tweet#17] Batched: false, DataFilters: [], Format: CSV, Location: InMemoryFileIndex(1 paths)[file:\/Users\/a616287\/Repos\/SBX\/files\/tweets.csv], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;userId:string,tweet:string><\/code><\/pre>\n<p>And with the usage of only standart <code>org.apache.spark.sql.functions._<\/code><\/p>\n<pre><code>== Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- Project [userId#16, word#41]    +- Filter (isnotnull(maxCount#57L) AND (count#52L = maxCount#57L))       +- Window [max(count#52L) windowspecdefinition(userId#16, specifiedwindowframe(RowFrame, unboundedpreceding$(), unboundedfollowing$())) AS maxCount#57L], [userId#16]          +- Sort [userId#16 ASC NULLS FIRST], false, 0             +- Exchange hashpartitioning(userId#16, 200), ENSURE_REQUIREMENTS, [id=#66]                +- HashAggregate(keys=[userId#16, word#41], functions=[count(word#41)])                   +- Exchange hashpartitioning(userId#16, word#41, 200), ENSURE_REQUIREMENTS, [id=#63]                      +- HashAggregate(keys=[userId#16, word#41], functions=[partial_count(word#41)])                         +- Generate explode(arrayOfWords#36), [userId#16], false, [word#41]                            +- Project [userId#16, split(tweet#17,  , -1) AS arrayOfWords#36]                               +- Filter ((size(split(tweet#17,  , -1), true) > 0) AND isnotnull(split(tweet#17,  , -1)))                                  +- FileScan csv [userId#16,tweet#17] Batched: false, DataFilters: [(size(split(tweet#17,  , -1), true) > 0), isnotnull(split(tweet#17,  , -1))], Format: CSV, Location: InMemoryFileIndex(1 paths)[file:\/Users\/a616287\/Repos\/SBX\/files\/tweets.csv], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;userId:string,tweet:string>ious.<\/code><\/pre>\n<p>I guess the conclusion regarding these two plans is obvious.<\/p>\n<p>So, <code>org.apache.spark.sql.expressions.Aggregator<\/code>  is a powerful abstraction in Spark SQL used for creating custom  aggregation functions that can be used in DataFrame API or SQL queries.  You should consider using <code>Aggregator<\/code> when:<\/p>\n<ol>\n<li>\n<p><strong>Custom Aggregation Logic<\/strong>: You need to perform custom aggregation operations that are not provided by built-in Spark SQL functions. <code>Aggregator<\/code> allows you to define your own aggregation logic .<\/p>\n<\/li>\n<li>\n<p><strong>Performance Optimization<\/strong>: Using <code>Aggregator<\/code> can provide performance benefits over attempting to perform aggregations using a combination of standard functions from <code>org.apache.spark.sql.functions._<\/code>. This is particularly advantageous compared to relying solely on UDFs or combinations of standard functions, as <code>Aggregator<\/code> leverages Spark&#8217;s Catalyst optimizer to generate optimized physical execution plans<\/p>\n<\/li>\n<li>\n<p><strong>Code Reusability<\/strong>: By encapsulating custom aggregation logic within an <code>Aggregator<\/code>, you can reuse the same logic across different parts of your codebase, making your code more modular and easier to maintain.<\/p>\n<\/li>\n<li>\n<p><strong>Integration with DataFrame API and SQL<\/strong>: <code>Aggregator<\/code>  seamlessly integrates with the DataFrame API and can be used directly  in SQL queries, allowing you to leverage the expressive power of SQL  while benefiting from the performance optimizations provided by <code>Aggregator<\/code>. (Not covered in this artical because I prefer DataFrame API :-))<\/p>\n<\/li>\n<\/ol>\n<p>Overall, <code>Aggregator<\/code>  is a powerful tool for building custom aggregation functions in Spark  SQL, especially when you need to perform complex aggregations  efficiently.<\/p>\n<\/p>\n<\/div>\n<\/div>\n<\/div>\n<p><!----><!----><\/div>\n<p><!----><!----><br \/> \u0441\u0441\u044b\u043b\u043a\u0430 \u043d\u0430 \u043e\u0440\u0438\u0433\u0438\u043d\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438 <a href=\"https:\/\/habr.com\/ru\/articles\/804625\/\"> https:\/\/habr.com\/ru\/articles\/804625\/<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"<div><!--[--><!--]--><\/div>\n<div id=\"post-content-body\">\n<div>\n<div class=\"article-formatted-body article-formatted-body article-formatted-body_version-2\">\n<div xmlns=\"http:\/\/www.w3.org\/1999\/xhtml\">\n<figure class=\"full-width\"><\/figure>\n<p>Below, we will discuss user-defined aggregation functions (UDAF) using <code>org.apache.spark.sql.expressions.Aggregator<\/code><strong>,<\/strong> which can be used for aggregating groups of elements in a DataSet into a single value in any user-defined way.<\/p>\n<p>Let\u2019s start by examining an example from the official documentation that implements a simple aggregation<\/p>\n<pre><code class=\"scala\">case class Data(i: Int)  val customSummer =  new Aggregator[Data, Int, Int] {      def zero: Int = 0      def reduce(b: Int, a: Data): Int = b + a.i      def merge(b1: Int, b2: Int): Int = b1 + b2      def finish(r: Int): Int = r      def bufferEncoder: Encoder[Int] = Encoders.scalaInt      def outputEncoder: Encoder[Int] = Encoders.scalaInt    }.toColumn()  val ds: Dataset[Data] = ... val aggregated = ds.select(customSummer)<\/code><\/pre>\n<p>The case class \u2018Data\u2019 is needed because UDAFs are used for DataSets.<\/p>\n<p>To create your own aggregator, you need to define 6 functions with predefined names:<\/p>\n<ul>\n<li>\n<p><strong>zero<\/strong> \u2014 also known as the initial value; should satisfy the requirement: \u201csomething\u201d + zero = \u201csomething\u201d<\/p>\n<\/li>\n<li>\n<p><strong>reduce<\/strong> \u2014 the reduce function that performs our aggregation into a buffer<\/p>\n<\/li>\n<li>\n<p><strong>merge<\/strong> \u2014 the function for merging buffers<\/p>\n<\/li>\n<li>\n<p><strong>finish<\/strong>  \u2014 the final processing to obtain the target value; in the specific  example, it is essentially absent, but it is often useful (as we\u2019ll see  in the examples below)<\/p>\n<\/li>\n<li>\n<p>two <strong>encoders<\/strong> for the buffer and the output value<\/p>\n<\/li>\n<\/ul>\n<p>Why  so many? Let\u2019s dive a little bit deeper. Spark is a distributed data  processing framework and processes aggregation in the following steps:<\/p>\n<ul>\n<li>\n<p><strong>First step:<\/strong> On the executors, we perform a pre-aggregation using a reduce function to minimize the volume of data shuffling.<\/p>\n<\/li>\n<li>\n<p><strong>Second step:<\/strong> Spark shuffles the data \u2014 moving the data with the same aggregation key to the same executor.<\/p>\n<\/li>\n<li>\n<p><strong>Third step: <\/strong>Finalizes aggregation on executors using a merge function.<\/p>\n<\/li>\n<li>\n<p><strong>Fourth step:<\/strong> Final processing using a finish function.<\/p>\n<\/li>\n<\/ul>\n<p>Therefore,  even if you don\u2019t need a certain step, we still have to define these  functions. On the other hand, you are unlikely to use UDAF for simple  summation, and for complex aggregations, such staging can be very  useful. Also, it is important to note that the aggregation is  essentially performed twice \u2014 first on the executors, and then the final  aggregation is done with data shuffling.<\/p>\n<p>Let\u2019s  consider an educational \u2018word count\u2019 example implemented using  aggregators. Suppose we have a simple CSV file with user tweets:<\/p>\n<pre><code>userId,tweet f6e8252f,cat dog frog cat dog frog frog dog dog bird f6e8252f,cat dog frog dog dog bird f6e8252f,cat dog 29d89ee4,frog frog dog bird 29d89ee4,frog cat dog frog frog dog dog bird 29d89ee4,frog bird<\/code><\/pre>\n<p>Let\u2019s  determine the most frequently used word for each user. Naturally, there  are many ways to solve this task, but for educational purposes, we will  demonstrate how to solve it using a UDAF.<\/p>\n<pre><code class=\"scala\">val df = spark.read.format(\"csv\").option(\"header\", \"true\").load(path)  type FrenquencyMapType = Map[String, Int]  class MyCustomAggregator(columnName: String) extends Aggregator [Row, FrenquencyMapType, String] {    private def combineMaps(map1: FrenquencyMapType, map2: FrenquencyMapType): FrenquencyMapType = {     val combinedKeys = map1.keySet ++ map2.keySet     combinedKeys.map { key => key -> (map1.getOrElse(key, 0) + map2.getOrElse(key, 0))     }.toMap   }    def zero: FrenquencyMapType = Map.empty[String, Int]    def reduce(buffer: FrenquencyMapType, row: Row): FrenquencyMapType = {     val wordsMap = row.getAs[Any](columnName).toString.split(\" \").groupBy(identity).mapValues(_.length)     combineMaps(buffer, wordsMap)    }    def merge(map1: FrenquencyMapType, map2: FrenquencyMapType): FrenquencyMapType =     combineMaps(map1, map2)    def finish(buffer: FrenquencyMapType): String =     if (buffer.isEmpty) \"nUll\" else buffer.maxBy(_._2)._1   def bufferEncoder: Encoder[FrenquencyMapType] = Encoders.kryo[Map[String, Int]]    def outputEncoder: Encoder[String] = Encoders.STRING  }  val mostPopularWordAggregator = new MyCustomAggregator(\"tweet\").toColumn  df.groupBy(col(\"userID\"))   .agg(mostPopularWordAggregator.name(\"favoriteWord\"))   .withColumnRenamed(\"value\", \"userId\")   .show()<\/code><\/pre>\n<p>The result:<\/p>\n<pre><code>+--------+---------------+ |userId  |favoriteWord   | +--------+---------------+ |29d89ee4|frog           | |f6e8252f|dog            | +--------+---------------+<\/code><\/pre>\n<p>Note the specification of types &lt;IN, BUF, OUT>:<\/p>\n<ul>\n<li>\n<p>The input type for our aggregator is Row.<\/p>\n<\/li>\n<li>\n<p>For the buffer, we use the type <code>Map[String, Int]<\/code>(alias <strong>FrenquencyMapType<\/strong>)<\/p>\n<\/li>\n<li>\n<p>Finally, the output type is simply String since the most popular word is returned.<\/p>\n<\/li>\n<\/ul>\n<p>Astute  readers may have noticed that, in essence, we adapted the \u2018most common  value\u2019 aggregator for our needs. Could we achieve the same result  without UDAF? Of course! One of the naives variants is:<\/p>\n<ol>\n<li>\n<p>Get an ArrayType column using <code>org.apache.spark.sql.functions.split<\/code><\/p>\n<\/li>\n<li>\n<p>Create a new row for each element in the array using <code>org.apache.spark.sql.functions.explode<\/code><\/p>\n<\/li>\n<li>\n<p>Perform standard aggregation over <strong>userId + word<\/strong>.<\/p>\n<\/li>\n<li>\n<p>Take the <strong>userId + word<\/strong> with max value of count-aggregation.<\/p>\n<\/li>\n<\/ol>\n<p>In the code it could look like this:<\/p>\n<pre><code class=\"scala\">val window = Window.partitionBy(\"userId\")  df.withColumn(\"arrayOfWords\", split(col(\"tweet\"), \" \"))   .withColumn(\"word\", explode(col(\"arrayOfWords\")))   .groupBy(\"userId\", \"word\")   .agg(count(\"word\").as(\"count\"))   .withColumn(\"maxCount\", max(\"count\").over(window))   .filter(col(\"count\") === col(\"maxCount\"))   .select(\"userId\", \"word\")   .show()<\/code><\/pre>\n<p>And now let\u2019s compare the plans. With UDAF it\u2019s very simple:<\/p>\n<pre><code>== Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- ObjectHashAggregate(keys=[userID#16], functions=[mycustomaggregator($line15.$read$$iw$$iw$$iw$$iw$MyCustomAggregator@69cf253c, Some(createexternalrow(userId#16.toString, tweet#17.toString, StructField(userId,StringType,true), StructField(tweet,StringType,true))), Some(interface org.apache.spark.sql.Row), Some(StructType(StructField(userId,StringType,true), StructField(tweet,StringType,true))), encodeusingserializer(input[0, java.lang.Object, true], true), decodeusingserializer(input[0, binary, true], scala.collection.immutable.Map, true), staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true, false), StringType, true, 0, 0)])    +- Exchange hashpartitioning(userID#16, 200), ENSURE_REQUIREMENTS, [id=#30]       +- ObjectHashAggregate(keys=[userID#16], functions=[partial_mycustomaggregator($line15.$read$$iw$$iw$$iw$$iw$MyCustomAggregator@69cf253c, Some(createexternalrow(userId#16.toString, tweet#17.toString, StructField(userId,StringType,true), StructField(tweet,StringType,true))), Some(interface org.apache.spark.sql.Row), Some(StructType(StructField(userId,StringType,true), StructField(tweet,StringType,true))), encodeusingserializer(input[0, java.lang.Object, true], true), decodeusingserializer(input[0, binary, true], scala.collection.immutable.Map, true), staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true, false), StringType, true, 0, 0)])          +- FileScan csv [userId#16,tweet#17] Batched: false, DataFilters: [], Format: CSV, Location: InMemoryFileIndex(1 paths)[file:\/Users\/a616287\/Repos\/SBX\/files\/tweets.csv], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;userId:string,tweet:string><\/code><\/pre>\n<p>And with the usage of only standart <code>org.apache.spark.sql.functions._<\/code><\/p>\n<pre><code>== Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- Project [userId#16, word#41]    +- Filter (isnotnull(maxCount#57L) AND (count#52L = maxCount#57L))       +- Window [max(count#52L) windowspecdefinition(userId#16, specifiedwindowframe(RowFrame, unboundedpreceding$(), unboundedfollowing$())) AS maxCount#57L], [userId#16]          +- Sort [userId#16 ASC NULLS FIRST], false, 0             +- Exchange hashpartitioning(userId#16, 200), ENSURE_REQUIREMENTS, [id=#66]                +- HashAggregate(keys=[userId#16, word#41], functions=[count(word#41)])                   +- Exchange hashpartitioning(userId#16, word#41, 200), ENSURE_REQUIREMENTS, [id=#63]                      +- HashAggregate(keys=[userId#16, word#41], functions=[partial_count(word#41)])                         +- Generate explode(arrayOfWords#36), [userId#16], false, [word#41]                            +- Project [userId#16, split(tweet#17,  , -1) AS arrayOfWords#36]                               +- Filter ((size(split(tweet#17,  , -1), true) > 0) AND isnotnull(split(tweet#17,  , -1)))                                  +- FileScan csv [userId#16,tweet#17] Batched: false, DataFilters: [(size(split(tweet#17,  , -1), true) > 0), isnotnull(split(tweet#17,  , -1))], Format: CSV, Location: InMemoryFileIndex(1 paths)[file:\/Users\/a616287\/Repos\/SBX\/files\/tweets.csv], PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;userId:string,tweet:string>ious.<\/code><\/pre>\n<p>I guess the conclusion regarding these two plans is obvious.<\/p>\n<p>So, <code>org.apache.spark.sql.expressions.Aggregator<\/code>  is a powerful abstraction in Spark SQL used for creating custom  aggregation functions that can be used in DataFrame API or SQL queries.  You should consider using <code>Aggregator<\/code> when:<\/p>\n<ol>\n<li>\n<p><strong>Custom Aggregation Logic<\/strong>: You need to perform custom aggregation operations that are not provided by built-in Spark SQL functions. <code>Aggregator<\/code> allows you to define your own aggregation logic .<\/p>\n<\/li>\n<li>\n<p><strong>Performance Optimization<\/strong>: Using <code>Aggregator<\/code> can provide performance benefits over attempting to perform aggregations using a combination of standard functions from <code>org.apache.spark.sql.functions._<\/code>. This is particularly advantageous compared to relying solely on UDFs or combinations of standard functions, as <code>Aggregator<\/code> leverages Spark&#8217;s Catalyst optimizer to generate optimized physical execution plans<\/p>\n<\/li>\n<li>\n<p><strong>Code Reusability<\/strong>: By encapsulating custom aggregation logic within an <code>Aggregator<\/code>, you can reuse the same logic across different parts of your codebase, making your code more modular and easier to maintain.<\/p>\n<\/li>\n<li>\n<p><strong>Integration with DataFrame API and SQL<\/strong>: <code>Aggregator<\/code>  seamlessly integrates with the DataFrame API and can be used directly  in SQL queries, allowing<\/p>\n<\/li>\n<\/ol>\n<\/div>\n<\/div>\n<\/div>\n<\/div>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[],"tags":[],"class_list":["post-372471","post","type-post","status-publish","format-standard","hentry"],"_links":{"self":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/372471","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=372471"}],"version-history":[{"count":0,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/372471\/revisions"}],"wp:attachment":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=372471"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=372471"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=372471"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}