title: PySpark: Operations with columns given different levels of aggregation and conditions
body: <p>I want to get a ratio of sentiment, and for that I need to calculate how many positives and how many negatives there are per topic, and then divide it by the total of records of each topic.</p>
  <p>Let's say I have this dataset:</p>
  <pre><code>+-----+---------+
  |topic|sentiment|
  +-----+---------+
  |Chair| positive|
  |Table| negative|
  |Chair| negative|
  |Chair| negative|
  |Table| positive|
  |Table| positive|
  |Table| positive|
  +-----+---------+
  </code></pre>
  <p>In this case, I could give a value of -1 to 'negative' and 1 to 'positive', then this ratio would be <code>0.5</code> in the case of Table <code>(negative + positive + positive + positive) / total_count)</code>, and <code>-0.33</code> in the case of Chair: <code>(positive + negative + negative) / total_count)</code>.</p>
  <p>I have come up with this solution but seems way too complicated:</p>
  <pre><code>from pyspark.sql import SparkSession
  from pyspark.sql.types import StructType, StructField, StringType
  from pyspark.sql.functions import col, when
  
  
  spark = SparkSession.builder.appName('SparkExample').getOrCreate()
  
  data_e = [(&quot;Chair&quot;,&quot;positive&quot;),
      (&quot;Table&quot;,&quot;negative&quot;),
      (&quot;Chair&quot;,&quot;negative&quot;),
      (&quot;Chair&quot;,&quot;negative&quot;),
      (&quot;Table&quot;,&quot;positive&quot;),
      (&quot;Table&quot;,&quot;positive&quot;)
    ]
  
  schema_e = StructType([ \
      StructField(&quot;topic&quot;,StringType(),True), \
      StructField(&quot;sentiment&quot;,StringType(),True), \
    ])
   
  df_e = spark.createDataFrame(data=data_e,schema=schema_e)
  
  df_e_int = df_e.withColumn('sentiment_int', 
                             when(col('sentiment') == 'positive', 1) \
                 .otherwise(-1)) \
                 .select('topic', 'sentiment_int')
  
  agg_e = df_e_int.groupBy('topic') \
                  .count() \
                  .select('topic', 
                          col('count').alias('counts'))
  
  agg_sum_e = df_e_int.groupBy('topic') \
                      .sum('sentiment_int') \
                      .select('topic', 
                              col('sum(sentiment_int)').alias('sum_value'))
  
  agg_joined_e = agg_e.join(agg_sum_e, 
                            agg_e.topic == agg_sum_e.topic, 
                            'inner') \
                      .select(agg_e.topic, 'counts', 'sum_value')
  
  final_agg_e = agg_joined_e.withColumn('sentiment_ratio', 
                                        (col('sum_value')/col('counts'))) \
                            .select('topic', 'sentiment_ratio')
  </code></pre>
  <p>The final output would look like this:</p>
  <pre><code>+-----+-------------------+
  |topic|    sentiment_ratio|
  +-----+-------------------+
  |Chair|-0.3333333333333333|
  |Table| 0.5               |
  +-----+-------------------+
  </code></pre>
  <p>What's the most efficient way of doing this?</p>