json - 如何将 kafka 消息值转换为特定模式？

Question

我正在尝试使用 Pyspark 从 Kafka 主题中读取数据。我想将该数据转换为特定的模式。但无法这样做。

这是我尝试过的：

>> df = spark.read.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "test1").load()
    
>> userSchema = StructType().add("Name", StringType(), True).add("Age", IntegerType(), True)

>> df1 = df.selectExpr("CAST(value AS STRING)")

>> df2 = df1.select(from_json(col("value"), userSchema))

>> df2.printSchema()
root
 |-- jsontostructs(value): struct (nullable = true)
 |    |-- Name: string (nullable = true)
 |    |-- Age: integer (nullable = true)

我想要的是：

>> df2.printSchema()
root
|-- Name: string (nullable = true)
|-- Age: integer (nullable = true)

有没有办法获得所需的架构？

Answer 1

对于面临同样问题的任何人，这是我实现这一目标的方法：

 df2 = df1.select(from_json(col("value"),userSchema)).select("jsontostructs(value).*")