我使用结构化数据流从Kafka流式传输数据 .readStream .option("kafka.bootstrap.serversCassandra表连接它
val d = df.select(from_json(col("value").cast("string"), schema).cast("string&qu
我有一个火花结构化流scala作业,它从kafka读取json消息并将数据写入S3。我配置了一个合流模式注册中心,该模式使用type=object的json格式。现在,我可以从注册表中检索模式,但是我需要在包含kafka记录的dataframe上使用这个模式。如何将valueRestResponseSchema转换为结构化类型,以便能够在from_json中应用?val值: Dat