How do I do functions.from_csv at spark structured stream

user_14060960 0
Viewed 685

I read lines from a kafka source and I want to build a kafka consumer... in spark structured streaming I know how to tell spark that the incoming lines are json type... how do I do the same with from_csv ?

   val lines = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("subscribe", "topic2")
      .option("startingOffsets", "earliest")
      .load()
      .selectExpr("CAST(value AS STRING)")
      .select(functions.from_json($"value", retailDataSchema).as("data"))
      lines.printSchema()

The schema is:

        val retailDataSchema = new StructType()
          .add("InvoiceNo", IntegerType)
          .add("Quantity", IntegerType)
          .add("Country", StringType)

Thank you!

The input data looks like this:

spark-structured-streaming apache-kafka apache-spark scala
1 Answers

You could do this work around:

    val lines = spark
            .readStream()
            .format("kafka")
            .option("kafka.bootstrap.servers", "localhost:9092")
            .option("subscribe", "topic2")
            .option("startingOffsets", "earliest")
            .load()
            .select(col("value").cast("string")).as("data").select("data.*").selectExpr("cast(split(value,',')[0] as DataTypes.IntegerType) as InvoiceNo"
                    ,"cast(split(value,',')[1] as DataTypes.IntegerType) as Quantity"
                    ,"cast(split(value,',')[2] as DataTypes.StringType) as Country" );
    lines.printSchema();

Or you could use the built-in function from_csv Since Apache spark 3.0.0

val lines = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "topic2")
  .option("startingOffsets", "earliest")
  .load()
  .selectExpr("CAST(value AS STRING)")
  .select(functions.from_csv($"value", retailDataSchema).as("data"))
  lines.printSchema()

Apache Spark Docs for from_csv built-in function

user_45431680
Related