azhur / kafka-serde-scala Goto Github PK

View Code? Open in Web Editor NEW

57.0 6.0 9.0 298 KB

Implicitly converts typeclass encoders to kafka Serializer, Deserializer, Serde.

Home Page: https://github.com/azhur/kafka-serde-scala

License: Apache License 2.0

Scala 100.00%

kafka kafka-streams json scala serialization kafka-serializers kafka-serde-scala serde deserialize json4s

kafka-serde-scala's Introduction

kafka-serde-scala

kafka-serde-scala provides implicit conversions from different type class Encoder/Decoder to kafka Serializer, Deserializer, Serde.

Following target libraries are supported:

Inspired by https://github.com/hseeberger/akka-http-json.

Installation

Add dependencies for the selected integration:

for avro4s:

libraryDependencies ++= List(
  "io.github.azhur" %% "kafka-serde-avro4s" % version
)

for circe:

libraryDependencies ++= List(
  "io.github.azhur" %% "kafka-serde-circe" % version
)

for jackson:

libraryDependencies ++= List(
  "io.github.azhur" %% "kafka-serde-jackson" % version
)

for json4s:

libraryDependencies ++= List(
  "io.github.azhur" %% "kafka-serde-json4s" % version
)

for jsoniter-scala:

libraryDependencies ++= List(
  "io.github.azhur" %% "kafka-serde-jsoniter-scala" % version,
  "com.github.plokhotnyuk.jsoniter-scala" %% "jsoniter-scala-macros" % jsoniter_scala_version % Provided // required only in compile-time
)

for play-json:

libraryDependencies ++= List(
  "io.github.azhur" %% "kafka-serde-play-json" % version
)

for upickle:

libraryDependencies ++= List(
  "io.github.azhur" %% "kafka-serde-upickle" % version
)

for scalapb:

libraryDependencies ++= List(
  "io.github.azhur" %% "kafka-serde-scalapb" % version
)

for zio-json:

libraryDependencies ++= List(
  "io.github.azhur" %% "kafka-serde-zio-json" % version
)

Usage

Mix xxxSupport into your code which requires implicit Kafka Serde, Serializer or Deserializer, where xxx is the target library used for serialization, i.e: CirceSupport.

Provide your implicit type class instances and the magic will convert them to Kafka serializers:

for avro4s: com.sksamuel.avro4s.SchemaFor[T], com.sksamuel.avro4s.ToRecord[T], com.sksamuel.avro4s.FromRecord[T]
for circe: io.circe.Encoder[T], io.circe.Decoder[T]
for jackson json: com.fasterxml.jackson.databind.ObjectMapper
for jackson binary: com.fasterxml.jackson.databind.ObjectMapper, org.codehaus.jackson.FormatSchema
for json4s: org.json4s.DefaultFormats, org.json4s.Serialization
for jsoniter-scala: com.github.plokhotnyuk.jsoniter_scala.core.JsonValueCodec[T], (and optionally com.github.plokhotnyuk.jsoniter_scala.core.WriterConfig or/and com.github.plokhotnyuk.jsoniter_scala.core.ReaderConfig)
for play-json: play.api.libs.json.Reads, play.api.libs.json.Writes
for upickle: upickle.default.Reader, upickle.default.Writer
for scalapb: scalapb.GeneratedMessageCompanion
for zio-json: zio.json.JsonEncoder, zio.json.JsonDecoder

For more info, please, take a look at unit tests and at kafka-serde-scala-example which is a kafka-streams (2.x) application with kafka-serde-scala usage.

Contribution

Feel free to contribute with creating PR or opening issues.

License

This code is open source software licensed under the Apache 2.0 License.

kafka-serde-scala's People

Contributors

Stargazers

Watchers

Forkers

gitter-badger ryanbreed-tp plokhotnyuk scala-steward adserver-bot sachabarber jiminhsieh lordlinus yanzzl

kafka-serde-scala's Issues

Support for serdes in windows?

I'm not sure if I'm missing something but would really appreciate some help. If I use no windows the implicits are found, for example:

// All good
object CustomerStream extends App with PlayJsonSupport {

  val totalPagesCounts: KTable[String, Long] = inputStream
    .filter((_ , ev) => ev.eventData.evData.pageType.isDefined && ev.eventData.custData.customerUid.isDefined)
    .groupBy((_, ev) => ev.eventData.custData.customerUid.get)
    //.windowedBy(SessionWindows.with(Duration.ofHours(3))
    .count()
  totalPagesCounts.toStream.to("total_pageviews")

}

but if I use windows, the produced implicits are not found (Produced[Windowed[String], Long]):

object CustomerStream extends App with PlayJsonSupport {
 // Error
  val totalPagesCounts: KTable[Windowed[String], Long] = inputStream
    .filter((_ , ev) => ev.eventData.evData.pageType.isDefined && ev.eventData.custData.customerUid.isDefined)
    .groupBy((_, ev) => ev.eventData.custData.customerUid.get)
    .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
    .count()
  totalPagesCounts.toStream.to("total_pageviews") \\ -> No implicits found for parameter produced: Produced[Windowed[String], Long]

}

Any idea why?

Thanks!

Add support for scala 2.13

Is it possible that we can get a compile going for a 2.13 version of this?

Can we get an example of a json4s Serdes?

The Circe example just auto imports everything and doesn't show what you would need to do in respect to other libraries.

Add avro schema registry support

Add support for binary formats from https://github.com/FasterXML/jackson-dataformats-binary

adding support to scalapb

Hello, I think this project is not JSON only specific, right? If possible, can we add support to scalapb? Thanks!

generic json

Is there a way to just encode the generic circe json object and send it ? Such as we have GenericAvro Record, or Generic Json from Gson see https://github.com/fabiojose/gson-serde, or again Generic Json with Jackson JsonNode from confluent https://github.com/axbaretto/kafka/blob/master/connect/json/src/main/java/org/apache/kafka/connect/json/JsonSerializer.java

How to use this with a Producer/Consumer (not KafkaStreams)

Hi,

I have been using the implicit Serdes for a KafkaStreams app very nicely, but for the integration tests, I also need to manually produce some messages.
As the producer actually requires a string I have been wondering how to achieve this.

The code snippet below shows the use case (which errors with a runtime error - not a build error)
Error: Invalid value com.example.reading_filter.models.JsonSerde$$anon$1@3aaf4f07 for configuration key.serializer: Expected a Class instance or class name

object KafkaHelper {
  private def fooSerializer()(implicit serde: Serde[Foo]) = serde.deserializer()


  def getProducer: KafkaProducer[Foo, Foo] = {
    val props = new Properties
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, appConfig.bootstrapServers)
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, fooSerializer)
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, fooSerializer)
    new KafkaProducer(props)
}

...
KafkaHelper.getProducer

I also tried fooSerializer.getClass, but this results in Could not find a public no-argument constructor for com.example.reading_filter.models.JsonSerde$$anon$1

Add AVSystem commons GenCodec support

https://github.com/AVSystem/scala-commons/blob/master/docs/GenCodec.md