Всем привет!

В этой статье мы посмотрим на два варианта решения "типичной" задачи потоковой обработки с учетом состояния при помощи фреймворка Apache Flink.

Если вы только начинаете знакомство с Flink, здесь можно найти обильное количество материалов.

Постановка задачи

Итак, давайте представим базу данных (пусть это будет PostgreSQL) состоящую из трех таблиц:

1. Clients - общая информация по клиенту:

2. ClientCompany - информация о компании клиента:

3. Payment - информация о платежах клиента:

Будем получать обновления по всем таблицам через CDC-канал (пусть это будет Debezium) в топик Kafka в формате Json. Задачей нашей Job-ы будет слушать топики и выдавать обновления по клиенту (например: клиент сменил компанию) во внешнюю систему (в рамках статьи в качестве внешней системы будем использовать консоль вывода).

Начинаем программировать

Наш технологический стек:

1. Scala v2.12.10

2. Flink v1.16.0

3. Circe v0.14.3

Мы разберем 2 варианта решения:

1. При помощи Table API

2. При помощи классического State Processor API

Прежде чем писать основную логику нам потребуется реализовать несколько общих шагов, которыми мы будем пользоваться в обеих реализациях.

Первое что нам потребуется - это научиться читать топики Kafka, в этом нам поможет уже готовый Flink Kafka Connector:

sealed trait KafkaConsumerSource {

  def configure[A: Decoder: ClassTag](bootstrapServers: String, topicName: String): KafkaSource[A] = {
    KafkaSource
      .builder()
      .setBootstrapServers(bootstrapServers)
      .setGroupId(s"$topicName-group")
      .setTopics(topicName)
      .setDeserializer(new KafkaJsonRecordDeserializer[A])
      .setStartingOffsets(OffsetsInitializer.earliest())
      .build()
  }
}

object KafkaConsumerSource extends KafkaConsumerSource {

  def configureKafkaDataSource[A: Decoder: ClassTag](
      topicName: String
  )(implicit env: StreamExecutionEnvironment,
    typeInformation: TypeInformation[A]): DataStream[A] = {
    val kafkaSource = KafkaConsumerSource.configure[A](Kafka.BootstrapServers, topicName)
    env.fromSource[A](kafkaSource, WatermarkStrategy.noWatermarks[A](), s"$topicName-flow")
  }
}

class KafkaJsonRecordDeserializer[A: Decoder: ClassTag] extends KafkaRecordDeserializationSchema[A] {

  override def deserialize(record: ConsumerRecord[Array[Byte], Array[Byte]], out: Collector[A]): Unit = {
    val value = new String(record.value())
    val obj   = Serde.toObj(value).toOption // в случае неудачи при десериализации, ошибка просто игнорируется и событие отбрасывается (в реальных проектах так делать не стоит)
    obj.foreach(out.collect)
  }

  override def getProducedType: TypeInformation[A] = {
    TypeExtractor
      .getForClass(classTag[A].runtimeClass.asInstanceOf[Class[A]])
      .asInstanceOf[TypeInformation[A]]
  }
}

В сериализации/десериализации сообщений нам поможет Circe:

object Serde {

  def toObj[A: Decoder](json: String): Either[circe.Error, A] = decode[A](json)

  def toJson[A: Encoder](obj: A): String = obj.asJson.toString()
}

Мы немного упростим себе задачу, сымитируем поведение CDC-канала, отправляя сообщения требуемого формата напрямую в топик Kafka:

case class Client (before: Option[Client.Value], after: Option[Client.Value], op: String)

object Client {
  case class Value(clientId: Int, name: String, surname: String, patronymic: Option[String], sex: String)
}

case class ClientCompany (before: Option[ClientCompany.Value], after: Option[ClientCompany.Value], op: String)

object ClientCompany {
  case class Value(clientId: Int, companyId: Int, companyName: String)
}

case class Payment (before: Option[Payment.Value], after: Option[Payment.Value], op: String)

object Payment {
  case class Value(clientId: Int, amount: Int, tmMs: Long)
}

Теперь мы готовы приступить к основной части.

Использование Flink Table API

Итак, прежде всего нам нужно получить необходимое окружение, поскольку мы будем работать в потоковом режиме, заручимся таковым:

implicit val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

Далее получим окружение для доступа к Table API :

implicit val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

Формируем Kafka Source при помощи кода который мы подготовили выше:

val clients: DataStream[Client] = KafkaConsumerSource.configureKafkaDataSource[Client]("clients")

Теперь нам остался всего один шаг до получения представления с которым нам предстоит работать - Dynamic Tables. Чтобы сделать этот шаг максимально простым, воспользуемся возможностями Scala и расширим DataStream методом toStreamTable:

implicit class DataStreamOps[A: RowTransformer](val stream: DataStream[A]) {

  def toStreamTable(primaryKey: String*)(implicit env: StreamTableEnvironment): Table = {
    val transformer = RowTransformer[A]
    stream
      .map(transformer.toRow(_))
      .flatMap(_.toSeq)(transformer.typeInformation)
      .toChangelogTable(
        env,
        Schema
          .newBuilder()
          .primaryKey(primaryKey: _*)
          .build(),
        ChangelogMode.upsert()
      )
  }
}

На вход функции подается список имен полей, которые составят первичный ключ нашей таблицы. Как и в большинстве фреймворков наименьшей единицей для работы с таблицами является строка (Row). Нужен способ преобразования всех наших сущностей в тип Row, продолжим пользоваться возможностями Scala, на этот раз тем что Scala позволяет писать код в функциональном стиле, а конкретно мы воспользуемся Type Class-ми. Наш Type Class будет иметь следующий вид:

sealed trait RowTransformer[A] {
  def toRow(entity: A): Option[Row]
  protected val fieldsName: Array[String]
  implicit def typeInformation: TypeInformation[Row]
}

Представим реализацию для одной из таблиц (для остальных алгоритм будет аналогичен):

implicit case object ClientRow extends RowTransformer[Client] {
  override def toRow(entity: Client): Option[Row] = {
    val kind   = kindOf(entity.operation)
    val record = entity.before.orElse(entity.after)
    record.map(r => Row.ofKind(kind, Int.box(r.clientId), r.name, r.surname, r.patronymic.orNull, r.sex))
  }

  override implicit def typeInformation: TypeInformation[Row] = {
    Types.ROW_NAMED(fieldsName, Types.INT, Types.STRING, Types.STRING, Types.STRING, Types.STRING)
  }

  override protected val fieldsName: Array[String] = Array("clientId", "name", "surname", "patronymic", "sex")
}

Теперь мы можем преобразовывать DataStream в Table одним действием:

val clients: Table = KafkaConsumerSource.configureKafkaDataSource[Client]("clients")
  .toStreamTable("clientId")

Это далеко не единственный способ создания таблиц, если мы изначально хотим работать с данными как с таблицей, необязательно сперва представлять данные в виде DataStream:

tableEnv.createTable(
  "companies",
  TableDescriptor
    .forConnector("kafka")
    .schema(
      Schema
        .newBuilder()
        .column("clientId", DataTypes.INT().notNull())
        .column("companyId", DataTypes.INT().notNull())
        .column("companyName", DataTypes.STRING().notNull())
        .primaryKey("clientId")
        .build()
    )
    .option(KafkaConnectorOptions.TOPIC, Seq("companies").asJava)
    .option(KafkaConnectorOptions.PROPS_BOOTSTRAP_SERVERS, Kafka.BootstrapServers)
    .option(KafkaConnectorOptions.PROPS_GROUP_ID, "companies-group")
    .option(KafkaConnectorOptions.SCAN_STARTUP_MODE, KafkaConnectorOptions.ScanStartupMode.EARLIEST_OFFSET)
    .option(KafkaConnectorOptions.VALUE_FORMAT, "debezium-json")
    .build()
)

val companies: Table = tableEnv.sqlQuery("SELECT * FROM companies")

companies.print("Companies")

И конечно же куда без SQL:

tableEnv.executeSql(
  """
    |CREATE TABLE payments (clientId INT NOT NULL, amount INT NOT NULL, tmMs BIGINT NOT NULL)
    |WITH (
    | 'connector' = 'kafka',
    | 'topic' = 'payments',
    | 'properties.bootstrap.servers' = 'localhost:9092',
    | 'properties.group.id' = 'payments-group',
    | 'scan.startup.mode' = 'earliest-offset',
    | 'format' = 'debezium-json'
    |)
    |""".stripMargin)

val payments: Table = tableEnv.sqlQuery("SELECT * FROM payments")

payments.print("Payments")

Мы окончательно подготовились к написанию основной логики. Для нашей задачи будем использовать inner join, т.е. при получении нового события мы будем выдавать результат во внешнюю систему только в случае наличия данных по ключу во всех таблицах.

clients
  .join(companies, $"client.clientId" === $"company.clientId")
  .join(payments, $"client.clientId" === $"payment.clientId")
  .select("client.clientId", "name", "surname", "companyId", "companyName", "amount", "timestamp")
  .toChangelogStream(Schema.derived(), ChangelogMode.upsert())
  .print("Portfolio")

Использование State Processor API

Снова начнем с окружения, в этом варианте нам будет достаточно только окружения для работы в потоковом режиме:

implicit val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

В прошлом варианте мы приводили все наши сущности к типу Table, сейчас нам нужно сделать нечто похожее:

case class PortfolioState(
   id: Int,
   client: Option[Client.Value],
   company: Option[ClientCompany.Value],
   payment: Option[Payment.Value]) {

  def complete: Option[Portfolio] = {
    for {
      client  <- client
      company <- company
      payment <- payment
    } yield {
      Portfolio(
        client.clientId,
        client.name,
        client.surname,
        company.companyId,
        company.companyName,
        payment.amount,
        payment.timestamp
      )
    }
  }
}

Мы описали свой собственный аналог таблицы, добавив в него метод complete формирующий результат только в случае наличия данных по всем трем сущностям (аналог inner join). Опишем способ трансформации исходной сущности в наш тип (для остальных сущностей будет аналогично):

object PortfolioState {

  def apply(client: Client): PortfolioState = {
    val id = client.before.map(_.clientId).getOrElse(client.after.map(_.clientId).get)
    PortfolioState(id, client.before.orElse(client.after), None, None)
  }
}

Итоговый результат:

val clients: DataStream[PortfolioState] = KafkaConsumerSource.configureKafkaDataSource[Client]("clients").map(PortfolioState(_))

Последнее чего нам не хватает - это обработчика событий, который будет поддерживать актуальное состояние и выдавать результаты:

class StateProcessor extends KeyedProcessFunction[Int, PortfolioState, Portfolio] {

  private var state: ValueState[PortfolioState] = _

  override def open(parameters: Configuration): Unit = {
    state = getRuntimeContext.getState(
      new ValueStateDescriptor[PortfolioState]("state", createTypeInformation[PortfolioState])
    )
  }

  override def processElement(
     value: PortfolioState,
     ctx: KeyedProcessFunction[Int, PortfolioState, Portfolio]#Context,
     ut: Collector[Portfolio]
  ): Unit = {
    Option(state.value()).fold {
      state.update(value) // если state == null, инициализируем его входным значением
    } { currentState =>
      val updatedState = currentState.copy(
        client = value.client.orElse(currentState.client),
        company = value.company.orElse(currentState.company),
        payment = value.payment.orElse(currentState.payment)
      )
      updatedState.complete.foreach(out.collect) // если данных хватает, то выдаем результат
      state.update(updatedState)
    }
  }
}

Соберем все вместе:

clients
  .union(companies) // объединяем все три потока в один
  .union(payments)
  .keyBy(_.id) // шардиурем данные по полю id (clientId)
  .process(new PortfolioStateProcessor())
  .print("Portfolio")

Итоги

Мы рассмотрели 2 разных подхода. Даже несмотря на всю простоту постановки, подход с применением Table API для задач отслеживания изменений и моментальной выдачи результата выглядит привлекательнее. Мы получаем возможность описывать желаемый результат на привычном SQL-подобном синтаксисе, находясь в бесконечном потоке событий.

Полный код проекта можно найти здесь