Spark потоковой передачи данных из темы Кафка и записи в текстовые файлы по внешнему пути

Question

Я хочу прочитать данные из темы и группы kafka по ключевым значениям и записать в текстовые файлы ..

public static void main(String[] args) throws Exception {
        SparkSession spark=SparkSession
                .builder()
                .appName("Sparkconsumer")
                .master("local[*]")
                .getOrCreate();
        SQLContext sqlContext = spark.sqlContext();
        SparkContext context = spark.sparkContext();
        Dataset<Row>lines=spark
                .readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers", "localhost:9092")
                .option("subscribe","test-topic")
                .load();
    Dataset<Row> r=  lines.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)");
        r.printSchema();
        r.createOrReplaceTempView("basicView");
        sqlContext.sql("select * from basicView")
        .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
        .writeStream()
        .outputMode("append")
        .format("console")
        .option("path","usr//path")
        .start()
        .awaitTermination();