title: Streaming JSON data from Kafka into Pyspark: aggregation not working on the JSON data
body: <p>I am trying to read JSON data from Kafka into Spark using Python and then doing some aggregation operations on the data, but there is some problem. I have the following code:
  First I just send the input data to the console:</p>
  <pre><code>import os
  os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0- 
  10_2.11:2.4.0 pyspark-shell'
  from pyspark.sql.functions import from_json
  import findspark
  findspark.init()
  
  from pyspark.sql import SparkSession
  from pyspark.sql.types import *
  from pyspark.sql.functions import col
  
  spark = SparkSession\
      .builder\
      .appName(&quot;Total-spending-for-top-users&quot;)\
      .getOrCreate()
  
  df = spark.readStream.format(&quot;kafka&quot;)\
      .option(&quot;kafka.bootstrap.servers&quot;, &quot;localhost:9092&quot;)\
      .option(&quot;subscribe&quot;, &quot;saurabmarjara2&quot;)\
      .option(&quot;startingOffsets&quot;, &quot;earliest&quot;)\
      .load()
  
  
  jsonschema = StructType([StructField(&quot;order_id&quot;, IntegerType()),
                  StructField(&quot;customer_id&quot;, IntegerType()),
                  StructField(&quot;taxful_total_price&quot;, IntegerType())])
  
  mta_stream = df.select(from_json(col(&quot;value&quot;).cast(&quot;string&quot;), jsonschema) \
                          .alias(&quot;parsed_mta_values&quot;))
  
  mta_data = mta_stream.select(&quot;parsed_mta_values.*&quot;)
  
  qry = mta_data.writeStream.outputMode(&quot;append&quot;).format(&quot;console&quot;).start()
  qry.awaitTermination()
  </code></pre>
  <h2>This works correctly. The output is:</h2>
  <div class="s-table-container">
  <table class="s-table">
  <thead>
  <tr>
  <th>order_id</th>
  <th>customer_id</th>
  <th>taxful_total_price</th>
  </tr>
  </thead>
  <tbody>
  <tr>
  <td>937665</td>
  <td>7</td>
  <td>10</td>
  </tr>
  <tr>
  <td>937666</td>
  <td>3</td>
  <td>4</td>
  </tr>
  <tr>
  <td>937667</td>
  <td>6</td>
  <td>3</td>
  </tr>
  <tr>
  <td>937668</td>
  <td>4</td>
  <td>4</td>
  </tr>
  </tbody>
  </table>
  </div>
  <p>But I want to perform aggregation on the data by grouping by the customer_id field and summing the taxful_total_price. This will give me the total spending for every customer.</p>
  <p>Here are the changes I made to the code:</p>
  <pre><code>df2 = 
  mta_data.groupBy(&quot;customer_id&quot;).agg(sum(&quot;taxful_total_price&quot;). 
  alias(&quot;total_spending&quot;))
  
  qry = df2.writeStream.outputMode(&quot;append&quot;).format(&quot;console&quot;).start()
  qry.awaitTermination()
  </code></pre>
  <p>This is the error I am getting:</p>
  <p>File &quot;sparkconsumer.py&quot;, line 31, in 
  df2=mta_data.groupBy(&quot;customer_id&quot;).agg(sum(&quot;taxful_total_price&quot;).alias(&quot;total_spending&quot;))
  TypeError: unsupported operand type(s) for +: 'int' and 'str'</p>
  <p>I have specified that each of the columns are integer types in the jsonschema. I think this is a problem with:</p>
  <pre><code>mta_stream.select(&quot;parsed_mta_values.*&quot;)
  </code></pre>
  <p>I tried this:</p>
  <pre><code> df2 = mta_data.groupBy( 
   &quot;parsed_mta_values.customer_id&quot;).agg(sum( 
   &quot;parsed_mta_values.taxful_total_price&quot;).alias(&quot;total_spending&quot;))
  </code></pre>
  <p>But this gives the same error as above.</p>
  <p>Please help!</p>