title: Is there anyway i can convert the Value data into actual column names in StructuredStreaming from Kafka?
body: <p>I have a csv file which has columns and for the testing purpose i push it manually to Kafka and from there i read it into Spark and apply some parsing and i do a console output for testing purposes. Now i understand the csv data is streamed as Value in structured streaming for which i cast it to String. My requirement is if i can convert the value data to the actual columns. There are hundreds of columns in the csv file but i am only looking at two specific columns "SERVICE_NAME8" &amp; "_raw"</p>
  
  <p>I use spark.sql to extract these columns when i read the csv file from a path but now i use structured streaming i am not sure if i can extract these specific columns as a new dataframe and apply my parsing thereafter</p>
  
  <pre><code>val df = spark
        .readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "10.160.172.45:9092, 10.160.172.46:9092, 10.160.172.100:9092")
        .option("subscribe", "TOPIC_WITH_COMP_P2_R2, TOPIC_WITH_COMP_P2_R2.DIT, TOPIC_WITHOUT_COMP_P2_R2.DIT")
        .load()
  
      val dfs = df.selectExpr("CAST(value AS STRING)").toDF()
  
      val data =dfs.withColumn("splitted", split($"value", "/"))
        .select($"splitted".getItem(4).alias("region"),$"splitted".getItem(5).alias("service"),col("value"))
        .withColumn("service_type", regexp_extract($"service", """.*(Inbound|Outbound|Outound).*""",1))
        .withColumn("region_type", concat(
          when(col("region").isNotNull,col("region")).otherwise(lit("null")), lit(" "),
          when(col("service").isNotNull,col("service_type")).otherwise(lit("null"))))
  
      val extractedDF = data.filter(
        col("region").isNotNull &amp;&amp;
          col("service").isNotNull &amp;&amp;
          col("value").isNotNull &amp;&amp;
          col("service_type").isNotNull &amp;&amp;
          col("region_type").isNotNull)
        .filter("region != ''")
        .filter("service != ''")
        .filter("value != ''")
        .filter("service_type != ''")
        .filter("region_type != ''")
  
  val query = extractedDF
  .writeStream
  .format("console")
  .outputMode("append")
  .trigger(ProcessingTime("20 seconds"))
  .start()
  </code></pre>
  
  <p>After val dfs = df.selectExpr("CAST(value AS STRING)").toDF() i somehow need to extract only two columns "SERVICE_NAME8" &amp; "_raw" and parsing should do the rest and produce output</p>