从kafka主题获取数据时如何拆分每行中的文本？

Question

我从一个 kafka 主题中检索数据。将数据转换为具有 10 列的数据框后，我选择其中一列。我想在每一行中拆分字符串，以便将单词转换为它们的发音。 evrything 似乎没问题，但我唯一的问题是我不能连续运行 split 方法。？

这是我的代码

   val df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "rocket-01.srvs.cloudkafka.com:9094")
            .option("subscribe", "k-news") 
            .option("startingOffsets", "latest")
            .option("kafka.security.protocol","SASL_SSL")
            .option("kafka.sasl.mechanism", "SCRAM-SHA-256")
            .option("kafka.sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=" " password="";").load()

val newsStringDF = df.selectExpr("CAST(value AS STRING)")

val newsSchema = StructType( Array(
     StructField("_c0",StringType,true),
     StructField("id",StringType,true),
     StructField("title",StringType,true),
     StructField("publication",StringType,true),
     StructField("author",StringType,true),
     StructField("date",StringType,true),
     StructField("year",StringType,true),
     StructField("month",StringType,true),
     StructField("url",StringType,true),
     StructField("content",StringType,true)))


val lines = Source.fromFile("/Project/symbols/cmudict.dict").getLines()
val res = lines.map { line =>line.split(" ", 2) match { case Array(a, b) => a -> b }}.toMap


val newsDF = newsStringDF.select(from_json(col("value"),
newsSchema).as("data")).select("data.*")
val titleColumn = newsDF.select("title").as[String].foreach( message => 
message.split(" ").toList.map { s =>if(res.get(s).isDefined) {res(s)} else 
{s}}.mkString(" ")).toDF("title") 

val streaming = titleColumn.writeStream.format("console").outputMode("append").trigger(Trigger.ProcessingTime("10 seconds")).start().awaitTermination()

输出应该是这样的： 我从我的 kafka-topic 获取消息（来自标题列的字符串），然后用另一个字符串替换它。 “我喜欢足球” --> “我喜欢 ˈfo͝otˌbôl” 并使用 .writeStream 将消息写入控制台

提前致谢

Answer 1

您必须记住，Dataframe 实际上只是 Dataset[Row]。当使用简单类型（甚至一些自定义案例类）时，很容易在有类型的Dataset 和无类型的Dataframe 之间进行转换。在这种特殊情况下，它应该很简单：

val titleColumn = newsDF
  .select("title")
  .as[String]
  .map( message => message.split(" ").toList.map { s =>
    if(res.get(s).isDefined) {res(s)} else {s}}.mkString(" ")
  )
  .toDF("title")

额外的行.as[String] 将您的Dataframe 转换为Dataset[String]，.toDF("title") 将其转换回来。这允许您的地图在Strings 而不是Rows 上运行。由于这是一个转换，您还需要在Dataset 上使用.map 而不是.foreach。

另一种选择是从Row 中检索标题String 并对其进行拆分：

message.getAs[String]("title").split(" ")

我不确定哪种方法更有效。这是一个经过测试的样本：

val df = Seq("aa 11", "bb 22", "cc 33", "dd 44").toDF("title")

val res = Map("aa" -> "AA", "bb" -> "BB", "cc" -> "CC", "dd" -> "DD")

df.as[String].map(m => {
  m.split(" ").toList.map(s => {
    if(res.get(s).isDefined) res(s) else s    
  }).mkString(" ")
}).toDF("title").show()

结果：

+-----+
|title|
+-----+
|AA 11|
|BB 22|
|CC 33|
|DD 44|
+-----+