How to read/write to Elasticsearch with Apache Spark with scala

Elastic Stack Elasticsearch
3

i'm trying to write and read to elasticsearch using apache spark
but data written to elasticsearch in format base64

this is the code i'm using to write to elasticsearch

var df = spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", KafkaService.bootstrapServers)
        .option("enable.auto.commit", KafkaService.enableAutoCommit)
        .option("failOnDataLoss", KafkaService.failOnDataLoss)
        .option("startingOffsets", KafkaService.startingOffsets)
        .option("subscribe", topicName)
        .option("group.id", groupId)
        .load()
    
    df.writeStream
    .outputMode(OutputMode.Append) //Only mode for ES
    .format("org.elasticsearch.spark.sql") //es
    .queryName("ElasticSink" + topicName)
    .start(indexName + "/broadcast") //ES index