Akka Flow如何对Kafka Topic消费的JSON数据应用filter过滤规则

核心逻辑 #

你需要先将Kafka消息携带的JSON字符串解析为可结构化访问的对象，再通过filter算子匹配Actor.Attributes.exit字段值为"0"的事件即可，注意你给出的JSON中exit是字符串类型，不是数值类型，不要写错匹配类型。

前置准备 #

首先定义和JSON结构匹配的样例类（也可直接用JSON库的动态API读取字段，无需映射样例类）：

case class Attributes(exit: String, `node.name`: String, name: String)
case class Actor(Attributes: Attributes, ID: String)
case class Event(Action: String, Actor: Actor)

调整后的Akka Flow代码 #

假设你已经实现了JSON反序列化方法parseEvent: KafkaMessage => Option[Event]，调整后的代码如下：

def getEvents: Flow[KafkaMessage, ConsumerMessage.CommittableOffset, NotUsed] =
    Flow[KafkaMessage]
      .via(getfromSource) //akka source 
      .wireTap(_.value.map(pprint.log(_))) // 已实现控制台打印所有Kafka Topic事件
      // 新增过滤逻辑
      .filter { msg =>
        parseEvent(msg)
          .map(_.Actor.Attributes.exit == "0")
          .getOrElse(false) // 解析失败的消息直接过滤，可根据业务需求调整
      }

如果用Play JSON的动态读取API，不需要定义样例类的写法如下：

import play.api.libs.json._
import scala.util.Try

def getEvents: Flow[KafkaMessage, ConsumerMessage.CommittableOffset, NotUsed] =
    Flow[KafkaMessage]
      .via(getfromSource)
      .wireTap(_.value.map(pprint.log(_)))
      .filter { msg =>
        // 请根据你实际的KafkaMessage结构调整获取JSON字符串的逻辑
        val jsonStr = msg.value.getOrElse("")
        Try(Json.parse(jsonStr)).toOption
          .flatMap(json => (json \ "Actor" \ "Attributes" \ "exit").asOpt[String])
          .contains("0")
      }

常见排查点 #

• 确认字段路径的大小写和JSON完全匹配，示例中的JSON用的是大写开头的Actor、Attributes，不要写成小写
• 确认你拿到的JSON字符串是完整合法的，没有格式错误
• 匹配时要用字符串"0"，不要用数值0，否则类型不匹配会一直过滤失败

内容的提问来源于stack exchange，提问作者curiousguy