我如何在 spark 结构化流中执行 functions.from_csv

Question

我从 kafka 源读取行，我想建立一个 kafka 消费者......在 spark 结构化流中 我知道如何告诉 spark 传入的行是 json 类型......我如何对 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()

架构是：

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

谢谢！

输入数据如下所示：

Answer 1

你可以解决这个问题：

    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();

或者您可以使用内置函数 from_csv 从 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