<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 TODO
Step1:  streamingInputDF.printSchema
```
root 
   |-- key
Step2:  Window
Step3:  Memory Output
Step4:  Console Output
Step5:  File Output with Partitions
Step6:  Create Table
Step7:  Add Partition
Step8:  Select
Step9:  Kafka Sink
<ASSISTANT_TASK:>
Python Code:
import org.apache.spark.sql.functions.get_json_object
import org.apache.spark.sql.functions.json_tuple
 
streamingInputDF = \
  spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "<server:ip>") \
    .option("subscribe", "topic1") \
    .option("startingOffsets", "latest") \
    .option("minPartitions", "10") \
    .option("failOnDataLoss", "true") \
    .load()
Explanation: TODO:  Convert to Python
Setup Connection to Kafka
End of explanation
import org.apache.spark.sql.functions._
var streamingSelectDF = 
  streamingInputDF
   .select(get_json_object(($"value").cast("string"), "$.zip").alias("zip"))
    .groupBy($"zip") 
    .count()
Explanation: streamingInputDF.printSchema
```
root 
   |-- key: binary (nullable = true) 
|-- value: binary (nullable = true) 
|-- topic: string (nullable = true) 
|-- partition: integer (nullable = true) 
|-- offset: long (nullable = true) 
|-- timestamp: timestamp (nullable = true) 
|-- timestampType: integer (nullable = true)
```
Sample Message
{
"city": "", 
"country": "United States", 
"countryCode": "US", 
"isp": "", 
"lat": 0.00, "lon": 0.00, 
"query": "", 
"region": "CA", 
"regionName": "California", 
"status": "success", 
"hittime": "2017-02-08T17:37:55-05:00", 
"zip": "38917" 
}
GroupBy, Count
End of explanation
import org.apache.spark.sql.functions._
var streamingSelectDF = 
  streamingInputDF
   .select(get_json_object(($"value").cast("string"), "$.zip").alias("zip"), get_json_object(($"value").cast("string"), "$.hittime").alias("hittime"))
   .groupBy($"zip", window($"hittime".cast("timestamp"), "10 minute", "5 minute", "2 minute"))
   .count()
Explanation: Window
End of explanation
import org.apache.spark.sql.streaming.ProcessingTime
val query =
  streamingSelectDF
    .writeStream
    .format("memory")        
    .queryName("isphits")     
    .outputMode("complete") 
    .trigger(ProcessingTime("25 seconds"))
    .start()
Explanation: Memory Output
End of explanation
import org.apache.spark.sql.streaming.ProcessingTime
val query =
  streamingSelectDF
    .writeStream
    .format("console")        
    .outputMode("complete") 
    .trigger(ProcessingTime("25 seconds"))
    .start()
Explanation: Console Output
End of explanation
import org.apache.spark.sql.functions._
var streamingSelectDF = 
  streamingInputDF
   .select(get_json_object(($"value").cast("string"), "$.zip").alias("zip"),    get_json_object(($"value").cast("string"), "$.hittime").alias("hittime"), date_format(get_json_object(($"value").cast("string"), "$.hittime"), "dd.MM.yyyy").alias("day"))
    .groupBy($"zip") 
    .count()
    .as[(String, String)]
Explanation: File Output with Partitions
End of explanation
%sql CREATE EXTERNAL TABLE  test_par
    (hittime string)
    PARTITIONED BY (zip string, day string)
    STORED AS PARQUET
    LOCATION '/mnt/sample/test-data'
Explanation: Create Table
End of explanation
%sql ALTER TABLE test_par ADD IF NOT EXISTS
    PARTITION (zip='38907', day='08.02.2017') LOCATION '/mnt/sample/test-data/zip=38907/day=08.02.2017'
Explanation: Add Partition
End of explanation
%sql select * from test_par
### JDBC Sink
import java.sql._
class  JDBCSink(url:String, user:String, pwd:String) extends ForeachWriter[(String, String)] {
      val driver = "com.mysql.jdbc.Driver"
      var connection:Connection = _
      var statement:Statement = _
      
    def open(partitionId: Long,version: Long): Boolean = {
        Class.forName(driver)
        connection = DriverManager.getConnection(url, user, pwd)
        statement = connection.createStatement
        true
      }
      def process(value: (String, String)): Unit = {
        statement.executeUpdate("INSERT INTO zip_test " + 
                "VALUES (" + value._1 + "," + value._2 + ")")
      }
      def close(errorOrNull: Throwable): Unit = {
        connection.close
      }
   }
val url="jdbc:mysql://<mysqlserver>:3306/test"
val user ="user"
val pwd = "pwd"
val writer = new JDBCSink(url,user, pwd)
val query =
  streamingSelectDF
    .writeStream
    .foreach(writer)
    .outputMode("update")
    .trigger(ProcessingTime("25 seconds"))
    .start()
Explanation: Select
End of explanation
import java.util.Properties
import kafkashaded.org.apache.kafka.clients.producer._
import org.apache.spark.sql.ForeachWriter
 class  KafkaSink(topic:String, servers:String) extends ForeachWriter[(String, String)] {
      val kafkaProperties = new Properties()
      kafkaProperties.put("bootstrap.servers", servers)
      kafkaProperties.put("key.serializer", "kafkashaded.org.apache.kafka.common.serialization.StringSerializer")
      kafkaProperties.put("value.serializer", "kafkashaded.org.apache.kafka.common.serialization.StringSerializer")
      val results = new scala.collection.mutable.HashMap[String, String]
      var producer: KafkaProducer[String, String] = _
      def open(partitionId: Long,version: Long): Boolean = {
        producer = new KafkaProducer(kafkaProperties)
        true
      }
      def process(value: (String, String)): Unit = {
          producer.send(new ProducerRecord(topic, value._1 + ":" + value._2))
      }
      def close(errorOrNull: Throwable): Unit = {
        producer.close()
      }
   }
val topic = "<topic2>"
val brokers = "<server:ip>"
val writer = new KafkaSink(topic, brokers)
val query =
  streamingSelectDF
    .writeStream
    .foreach(writer)
    .outputMode("update")
    .trigger(ProcessingTime("25 seconds"))
    .start()
Explanation: Kafka Sink
End of explanation
<END_TASK>