Spark : writeStream' can be called only on streaming Dataset/DataFrame

I'm trying to retrieve tweets from my Kafka cluster to Spark Streaming in which I perform some analysis to store them in an ElasticSearch Index.

Versions : Spark - 2.3.0 Pyspark - 2.3.0 Kafka - 2.3.0 Elastic Search - 7.9 Elastic Search Hadoop - 7.6.2

I run the following code in my Jupyter env to write the streaming dataframe into Elastic Search .

import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.3.0,org.elasticsearch:elasticsearch-hadoop:7.6.2 pyspark-shell'

from pyspark import SparkContext
#    Spark Streaming
from pyspark.streaming import StreamingContext
#    Kafka
from pyspark.streaming.kafka import KafkaUtils
#    json parsing
import json
import nltk
import logging
from datetime import datetime
from pyspark.sql import *
from pyspark.sql.types import *
from pyspark.sql.functions import *
from nltk.sentiment.vader import SentimentIntensityAnalyzer

def getSqlContextInstance(sparkContext):
    if ('sqlContextSingletonInstance' not in globals()):
        globals()['sqlContextSingletonInstance'] = SQLContext(sparkContext)
    return globals()['sqlContextSingletonInstance']


def analyze_sentiment(tweet):

    scores = dict([('pos', 0), ('neu', 0), ('neg', 0), ('compound', 0)])
    sentiment_analyzer = SentimentIntensityAnalyzer()
    score = sentiment_analyzer.polarity_scores(tweet)
    for k in sorted(score):
        scores[k] += score[k]

    return json.dumps(scores)

def process(time,rdd):
    
     print("========= %s =========" % str(time))
     
     try:
        if rdd.count()==0: 
            raise Exception('Empty')
            
        sqlContext = getSqlContextInstance(rdd.context)
        
        df = sqlContext.read.json(rdd)
        df = df.filter("text not like 'RT @%'")
        
        if df.count() == 0: 
            raise Exception('Empty')
            
        udf_func = udf(lambda x: analyze_sentiment(x),returnType=StringType())
        df = df.withColumn("Sentiment",lit(udf_func(df.text)))
        print(df.take(10))
        
        
        df.writeStream.outputMode('append').format('org.elasticsearch.spark.sql').option('es.nodes','localhost').option('es.port',9200)\
        .option('checkpointLocation','/checkpoint').option('es.spark.sql.streaming.sink.log.enabled',False).start('PythonSparkStreamingKafka_RM_01').awaitTermination()
        
        
     except Exception as e:
        print(e)
        pass

sc = SparkContext(appName="PythonSparkStreamingKafka_RM_01")
sc.setLogLevel("INFO")

ssc = StreamingContext(sc, 20)

kafkaStream = KafkaUtils.createDirectStream(ssc, ['kafkaspark'], {
                        'bootstrap.servers':'localhost:9092', 
                        'group.id':'spark-streaming', 
                        'fetch.message.max.bytes':'15728640',
                        'auto.offset.reset':'largest'})

parsed = kafkaStream.map(lambda v: json.loads(v[1]))

parsed.foreachRDD(process)

ssc.start()
ssc.awaitTermination(timeout=180)

But I get the error :

'writeStream' can be called only on streaming Dataset/DataFrame;

And , it looks like I have to use .readStream , but how do I use it to read from KafkaStream without CreateDirectStream ?

Could someone please help me with writing this dataframe into Elastic Search . I am a beginner to Spark Streaming and Elastic Search and find it quite challenging . Would be happy if someone could guide me through getting this done.