title: How to map kafka consumer data to Mysql using python
body: <p>My host configuaration is as follows:</p>
  <p>Kafka,
  Spark,
  Mysql ,
  are on docker</p>
  <p>My code is as follows:</p>
  <pre><code># To find out where the pyspark
  import sys
  from kafka import KafkaProducer,KafkaConsumer
  import findspark
  from boto3 import *
  import boto3
  import json
  
  findspark.init()
  # Creating Spark Context
  from pyspark import SparkContext
  from pyspark.sql import SparkSession
  def get_connection(self):
       spark = SparkSession.builder.master(&quot;local[*]&quot;).appName(&quot;SparkByExamples.com&quot;).getOrCreate()  
       return spark   
  
  def json_serializer(data):
       return json.dumps(data).encode(&quot;utf-8&quot;)
      
  
  def read_s3():
      p1 = KafkaProducer(bootstrap_servers=['broker:29092'], value_serializer=json_serializer)
      
      s3 = boto3.resource('s3')
      bucket = s3.Bucket('kakfa')
      for obj in bucket.objects.all():
          key = obj.key
          body = obj.get()['Body'].read().decode('utf-8')
      p1.send(&quot;Uber_Eats&quot;,body)
      p1.flush()
  def read_from_topic(self,spark):
      df = spark.readStream \
          .format(&quot;kafka&quot;) \
          .option(&quot;kafka.bootstrap.servers&quot;, &quot;broker:29092&quot;) \
          .option(&quot;subscribe&quot;, &quot;Uber_Eats&quot;) \
           .option(&quot;startingOffsets&quot;, &quot;earliest&quot;) \
          .load()
      df2 = df \
          .writeStream \
          .format(&quot;console&quot;) \
          .start()
      print(df2.awaitTermination()  )  
  def get_consumer(self):
      consumer = KafkaConsumer(&quot;Uber_Eats&quot;, group_id='group1', bootstrap_servers=
      &quot;broker:29092&quot;,value_deserializer=lambda x: json.loads(x.decode('utf-8')))
      return  consumer   
  def print_details(self,c1):
      for msg in c1:
            print(msg.value)
      print(&quot;Dom  dfe&quot;)            
             
            
  class Foo:
      def __init__(self):
                   
          spark = get_connection(self)
          read_s3()
          # System.setProperty(&quot;hadoop.home.dir&quot;, &quot;$HADOOP_HOME\winutils-master\hadoop-2.x.x&quot;)
          c1 = get_consumer(spark)
          print_details(self,c1)
  
  
  f = Foo()  
  </code></pre>
  <p>My output from above code is as folows:</p>
  <pre><code>{
      
          {
              &quot;Customer Number&quot;: &quot;1&quot;,
              &quot;Customer Name&quot;: &quot;Shyam&quot;,
              &quot;Restaurant Number&quot;: &quot;2201&quot;,
              &quot;Restaurant NameOrdered&quot;: &quot;Bawarchi&quot;,
              &quot;Number of Items&quot;: &quot;3&quot;,
              &quot;price&quot;: &quot;10&quot;,
              &quot;Operating Start hours&quot;: &quot;9:00&quot;,
              &quot;Operating End hours&quot;: &quot;23:00&quot;
          },
          {
              &quot;Customer Number&quot;: &quot;2&quot;,
              &quot;Customer Name&quot;: &quot;Rohini&quot;,
              &quot;Restaurant Number&quot;: &quot;2202&quot;,
              &quot;Restaurant NameOrdered&quot;: &quot;Sarvana Bhavan&quot;,
              &quot;Number of Items&quot;: &quot;4&quot;,
              &quot;price&quot;: &quot;20&quot;,
              &quot;Operating Start hours&quot;: &quot;8:00&quot;,
              &quot;Operating End hours&quot;: &quot;20:00&quot;
          },
          {
              &quot;Customer Number&quot;: &quot;3&quot;,
              &quot;Customer Name&quot;: &quot;Bhairav&quot;,
              &quot;Restaurant Number&quot;: &quot;2203&quot;,
              &quot;Restaurant NameOrdered&quot;: &quot;Taco Bell&quot;,
              &quot;Number of Items&quot;: &quot;5&quot;,
              &quot;price&quot;: &quot;30&quot;,
              &quot;Operating Start hours&quot;: &quot;11:00&quot;,
              &quot;Operating End hours&quot;: &quot;21:00&quot;
          }
      
  }
  </code></pre>
  <p>How do I read this to columns into mysql?
  i) Is it like a regular json file , read and insert?</p>
  <p>ii)Or Do we have anything kakfa consumer  'json' format specific?</p>
  <p>iii) I have specified value_deserializer=lambda x: json.loads(x.decode('utf-8')))</p>
  <p>in code to get in json format is this necessary to load data into mysql</p>
  <p>Thanks,</p>
  <p>Adi</p>