title: Optimized way to find values with top 20 frequencies in spark dataframe
body: <p>We have a spark dataframe. We are trying to find the values with top 20 frequencies in a column.
  Ex) [1, 1, 1, 2, 2, 4]</p>
  <p>In the above list,
  1 is occuring 3 times
  2 is occuring 2 times
  4 is occuring 1 time</p>
  <p>We are trying to find this using pandas.</p>
  <p>And then creating UDF in spark and using it there.
  This works for smaller datasets, but when the datasets are too tall (20M rows), we are facing memory issues sometimes.</p>
  <pre class="lang-py prettyprint-override"><code>from pyspark.sql import functions as F
  import pandas as pd
  from pyspark.sql.types import *
  from pyspark.sql import SparkSession
  spark = SparkSession.builder.appName(&quot;unit_testing_spark&quot;).getOrCreate()
  spark.sparkContext.setLogLevel(&quot;WARN&quot;)
  
  
  def find_freq_values(col_list):
      if len(col_list) == 0:
          return []
  
      df = pd.DataFrame(col_list, columns=[&quot;value&quot;])
      df = df[['value']].groupby(['value'])['value'] \
          .count() \
          .reset_index(name='count') \
          .sort_values(['count'], ascending=False) \
          .head(20)
      res = df.to_dict(orient='records')
  
      for curr_data in res:
          curr_value = curr_data[&quot;value&quot;]
          ldt = str(type(curr_value)).lower()
          if &quot;time&quot; in ldt or &quot;date&quot; in ldt:
              curr_data[&quot;value&quot;] = str(curr_value)
  
      return res
  
  s = find_freq_values([1, 1, 1, 2, 2, 4])
  print(s)
  # Output: [{'value': 1, 'count': 3}, {'value': 2, 'count': 2}, {'value': 4, 'count': 1}]
  
  column_data = [&quot;col_1&quot;]
  column_header = tuple(column_data)
  
  data = [[1], [1], [1], [2], [2], [4]]
  df = spark.createDataFrame(data, column_header)
  
  find_freq_udf = F.udf(find_freq_values, ArrayType(MapType(StringType(), StringType(), True)))
  
  freq_res_df = df.select(*[find_freq_udf(F.collect_list(c)).alias(c) for c in df.columns])
  freq_res = freq_res_df.collect()[0].asDict()
  print(freq_res)
  # Output: {'col_1': [{'count': '3', 'value': '1'}, {'count': '2', 'value': '2'}, {'count': '1', 'value': '4'}]}
  
  </code></pre>
  <p>Error message:
  &quot;An error occurred while calling o514.collectToPython.\n: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 66.0 failed 10 times, most recent failure: Lost task 0.9 in stage 66.0 (TID 379) (w382f6d7a114442a8bd741d53661a2c3b-srpz3ldim3sr2-w-1.c.projectid.internal executor 10): ExecutorLostFailure (executor 10 exited caused by one of the running tasks) Reason: Container from a bad node: container_1658234225716_0001_01_000011 on host: w382f6d7a114442a8bd741d53661a2c3b-srpz3ldim3sr2-w-1.c.projectid.internal. Exit status: 143.</p>
  <p>To perform multiple columns parallely we are using the following statement.</p>
  <p>How can this be optimized to avoid memory issue?</p>