如何实现基于Case Class与Schema的通用Kafka Struct生成方法？

我有多个Case Class/模型，需要为它们自动创建Kafka Struct。目前我通过以下代码手动实现该功能，但每个Case Class字段不同，都需要手动编写逻辑。是否存在一种通用方法，可接收Case Class、Schema和记录并生成对应的Kafka Struct？

注：xyz为包含hitDateTime、marketplaceId、hitYear和hitMonth字段的Case Class/模型。

override val KAFKA_SCHEMA: Schema = xyz.KAFKA_SCHEMA
override def buildKafkaStruct(record: Any): Struct = {
  val parsedRecord = record.asInstanceOf[xyz]
  val structRecord = new Struct(KAFKA_SCHEMA)
    .put("time", parsedRecord.time)
    .put("marketplace", parsedRecord.marketplace)

  structRecord
}
}

当然有通用方法，利用Scala的反射或宏机制就能实现自动映射，不用为每个Case Class重复编写赋值逻辑。以下是两种实用的实现方式：

1. 基于Scala反射的通用工具类 #

通过反射获取Case Class的字段信息，匹配Kafka Schema的字段名完成自动赋值，还支持字段名映射处理：

import org.apache.kafka.connect.data.Struct
import org.apache.kafka.connect.data.Schema
import scala.reflect.runtime.universe._

object KafkaStructBuilder {
  // 泛型方法，接收任意Case Class、Schema，返回对应的Kafka Struct
  def build[T: TypeTag](record: T, schema: Schema): Struct = {
    val struct = new Struct(schema)
    val mirror = runtimeMirror(record.getClass.getClassLoader)
    val instanceMirror = mirror.reflect(record)
    val classSymbol = typeTag[T].tpe.typeSymbol.asClass
    val constructorSymbol = classSymbol.primaryConstructor.asMethod

    // 遍历Case Class的所有构造参数（即字段）
    constructorSymbol.paramLists.flatten.foreach { param =>
      val caseFieldName = param.name.toString
      // 处理字段名不一致的情况，比如Case Class的hitDateTime对应Schema的time
      val schemaFieldName = caseFieldName match {
        case "hitDateTime" => "time"
        case "marketplaceId" => "marketplace"
        case other => other
      }

      // 如果Schema中存在该字段，则赋值
      Option(schema.field(schemaFieldName)).foreach { field =>
        val fieldMirror = instanceMirror.reflectField(param)
        struct.put(schemaFieldName, fieldMirror.get)
      }
    }
    struct
  }
}

使用示例 #

// 针对xyz类型的实例调用
val xyzRecord: xyz = xyz(hitDateTime = 1690000000L, marketplaceId = "US", hitYear = 2024, hitMonth = 7)
val kafkaStruct = KafkaStructBuilder.build(xyzRecord, xyz.KAFKA_SCHEMA)

2. 利用Magnolia宏库实现编译期映射 #

如果追求性能（反射有运行时开销），可以用Magnolia这类Scala宏库，在编译期自动生成映射逻辑，运行时无额外损耗：

首先添加sbt依赖：

libraryDependencies += "com.propensive" %% "magnolia" % "1.1.0"

然后定义类型类和自动派生逻辑：

import org.apache.kafka.connect.data.Struct
import org.apache.kafka.connect.data.Schema
import magnolia._

// 定义类型类，抽象Case Class到Struct的映射能力
trait KafkaStructMapper[T] {
  def toStruct(record: T, schema: Schema): Struct
}

object KafkaStructMapper {
  type Typeclass[T] = KafkaStructMapper[T]

  // 为Case Class自动生成映射逻辑
  def combine[T](ctx: CaseClass[KafkaStructMapper, T]): KafkaStructMapper[T] = new KafkaStructMapper[T] {
    override def toStruct(record: T, schema: Schema): Struct = {
      val struct = new Struct(schema)
      ctx.parameters.foreach { param =>
        val caseFieldName = param.label
        val schemaFieldName = caseFieldName match {
          case "hitDateTime" => "time"
          case "marketplaceId" => "marketplace"
          case other => other
        }

        Option(schema.field(schemaFieldName)).foreach { field =>
          // 递归处理嵌套类型（如果有）
          val value = param.typeclass.toStruct(param.dereference(record), field.schema())
          struct.put(schemaFieldName, value)
        }
      }
      struct
    }
  }

  // 为基础类型提供默认实现
  implicit val stringMapper: KafkaStructMapper[String] = (value, _) => value
  implicit val intMapper: KafkaStructMapper[Int] = (value, _) => value
  implicit val longMapper: KafkaStructMapper[Long] = (value, _) => value
  // 根据需要添加其他基础类型（如Double、Boolean等）

  // 宏方法，自动派生任意Case Class的映射实例
  def derive[T]: KafkaStructMapper[T] = macro Magnolia.gen[T]
}

使用示例 #

import KafkaStructMapper.derive

// 自动生成xyz类型的映射器
implicit val xyzMapper: KafkaStructMapper[xyz] = derive[xyz]

val xyzRecord: xyz = xyz(hitDateTime = 1690000000L, marketplaceId = "US", hitYear = 2024, hitMonth = 7)
val kafkaStruct = xyzMapper.toStruct(xyzRecord, xyz.KAFKA_SCHEMA)

关键注意点 #

• 字段名映射：如果Case Class和Schema的字段名不一致，可通过match分支添加自定义映射规则，也可以扩展为注解方式标记对应关系
• 类型兼容性：确保Case Class字段类型与Kafka Schema定义的类型匹配，比如Scala Long对应Kafka INT64，String对应STRING
• 性能选择：反射实现简单但有运行时开销，适合低吞吐量场景；宏生成的方式编译期完成逻辑，性能更优，适合高吞吐量场景

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