我很难用spark结构化流媒体读取kafka主题中的json数据。
上下文:
我正在构建一个简单的管道,在这里我使用kafka从mongodb读取数据(这个db经常从另一个应用程序填充),然后我想在spark中获取这些数据。
为此,我使用spark结构化流媒体,这似乎是可行的。
这是我的密码:
import org.apache.spark.rdd
import org.apache.spark.sql.avro._
import org.apache.spark.sql.{Column, SparkSession}
import org.apache.spark.sql.functions.from_json
import org.apache.spark.sql.types.{ArrayType, DataTypes, StructType}
import org.apache.spark.sql.functions.explode
import org.apache.spark.sql.functions.schema_of_json
object KafkaToParquetLbcAutomation extends App {
val spark = SparkSession
.builder
.appName("Kafka-Parquet-Writer")
.master("local")
.getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
import spark.implicits._
val kafkaRawDf = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers",BROKER IP)
.option("subscribe", "test")
.option("startingOffsets", "earliest")
.load()
val testJsonDf = kafkaRawDf.selectExpr("CAST(value AS STRING)")
//affichage des data
val query = testJsonDf
.writeStream
.outputMode("append")
.format("console")
.queryName("test")
.start()
.awaitTermination()
}
在阅读了这些json数据之后,我想进行一些转换。
这里开始的问题,我不能解析的json数据,由于一个奇怪的编码,我无法解码。
因此,我不能再继续我的管道。
如何获取数据:
{
"field 1" : "value 1 ",
}
(与许多其他领域)
如何实际获取数据:
VoituresXhttps://URL.fr/voitures/87478648654.htm�https://img5.url.fr/ad-image/49b7c279087d0cce09123a66557b71d09c01a6d2.jpg�https://img7.url.fr/ad-image/eab7e65419c17542840204fa529b02e64771adbb.jpg�https://img7.urln.fr/ad-image/701b547690e48f11a6e0a1a9e72811cc76fe803e.jpg
问题可能出在分隔符之类的地方。
你能帮帮我吗
谢谢您
1条答案
按热度按时间xzv2uavs1#
解决问题,
这是Kafka连接器代码中的错误配置。
我只需将此字段添加到连接器:
与spark无关