{"id":349722,"date":"2023-07-04T09:04:43","date_gmt":"2023-07-04T09:04:43","guid":{"rendered":"http:\/\/savepearlharbor.com\/?p=349722"},"modified":"-0001-11-30T00:00:00","modified_gmt":"-0001-11-29T21:00:00","slug":"","status":"publish","type":"post","link":"https:\/\/savepearlharbor.com\/?p=349722","title":{"rendered":"\u041a\u0430\u043a Flink Table API \u0443\u043f\u0440\u043e\u0449\u0430\u0435\u0442 \u0440\u0430\u0437\u0440\u0430\u0431\u043e\u0442\u043a\u0443<\/span>"},"content":{"rendered":"
<\/div>\n
\n
\n
\n
\n
<\/figure>\n

\u0412\u0441\u0435\u043c \u043f\u0440\u0438\u0432\u0435\u0442!<\/p>\n

\u0412 \u044d\u0442\u043e\u0439 \u0441\u0442\u0430\u0442\u044c\u0435 \u043c\u044b \u043f\u043e\u0441\u043c\u043e\u0442\u0440\u0438\u043c \u043d\u0430 \u0434\u0432\u0430 \u0432\u0430\u0440\u0438\u0430\u043d\u0442\u0430 \u0440\u0435\u0448\u0435\u043d\u0438\u044f «\u0442\u0438\u043f\u0438\u0447\u043d\u043e\u0439» \u0437\u0430\u0434\u0430\u0447\u0438 \u043f\u043e\u0442\u043e\u043a\u043e\u0432\u043e\u0439 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 \u0441 \u0443\u0447\u0435\u0442\u043e\u043c \u0441\u043e\u0441\u0442\u043e\u044f\u043d\u0438\u044f \u043f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a\u0430 Apache Flink<\/em>.<\/p>\n

\u0415\u0441\u043b\u0438 \u0432\u044b \u0442\u043e\u043b\u044c\u043a\u043e \u043d\u0430\u0447\u0438\u043d\u0430\u0435\u0442\u0435 \u0437\u043d\u0430\u043a\u043e\u043c\u0441\u0442\u0432\u043e \u0441 Flink<\/em>, \u0437\u0434\u0435\u0441\u044c<\/a> \u043c\u043e\u0436\u043d\u043e \u043d\u0430\u0439\u0442\u0438 \u043e\u0431\u0438\u043b\u044c\u043d\u043e\u0435 \u043a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u043e \u043c\u0430\u0442\u0435\u0440\u0438\u0430\u043b\u043e\u0432.<\/p>\n

\u041f\u043e\u0441\u0442\u0430\u043d\u043e\u0432\u043a\u0430 \u0437\u0430\u0434\u0430\u0447\u0438<\/h2>\n

\u0418\u0442\u0430\u043a, \u0434\u0430\u0432\u0430\u0439\u0442\u0435 \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u0438\u043c \u0431\u0430\u0437\u0443 \u0434\u0430\u043d\u043d\u044b\u0445 (\u043f\u0443\u0441\u0442\u044c \u044d\u0442\u043e \u0431\u0443\u0434\u0435\u0442 PostgreSQL<\/em>) \u0441\u043e\u0441\u0442\u043e\u044f\u0449\u0443\u044e \u0438\u0437 \u0442\u0440\u0435\u0445 \u0442\u0430\u0431\u043b\u0438\u0446:<\/p>\n

    \n
  1. \n

    Clients<\/em> — \u043e\u0431\u0449\u0430\u044f \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044f \u043f\u043e \u043a\u043b\u0438\u0435\u043d\u0442\u0443:<\/p>\n<\/li>\n<\/ol>\n

    \n
    \n\n\n\n\n\n\n\n\n
    \n

    \u0418\u043c\u044f \u043f\u043e\u043b\u044f<\/p>\n<\/th>\n

    		\n

    \u0422\u0438\u043f \u043f\u043e\u043b\u044f<\/p>\n<\/th>\n

    		\n

    \u041e\u043f\u0438\u0441\u0430\u043d\u0438\u0435<\/p>\n<\/th>\n<\/tr>\n

    \n

    clientId<\/p>\n<\/td>\n

    		\n

    int<\/p>\n<\/td>\n

    		\n

    \u0443\u043d\u0438\u043a\u0430\u043b\u044c\u043d\u044b\u0439 \u0438\u0434\u0435\u043d\u0442\u0438\u0444\u0438\u043a\u0430\u0442\u043e\u0440 \u043a\u043b\u0438\u0435\u043d\u0442\u0430<\/p>\n<\/td>\n<\/tr>\n

    \n

    name<\/p>\n<\/td>\n

    		\n

    string<\/p>\n<\/td>\n

    		\n

    \u0438\u043c\u044f \u043a\u043b\u0438\u0435\u043d\u0442\u0430<\/p>\n<\/td>\n<\/tr>\n

    \n

    surname<\/p>\n<\/td>\n

    		\n

    string<\/p>\n<\/td>\n

    		\n

    \u0444\u0430\u043c\u0438\u043b\u0438\u044f \u043a\u043b\u0438\u0435\u043d\u0442\u0430<\/p>\n<\/td>\n<\/tr>\n

    \n

    patronymic<\/p>\n<\/td>\n

    		\n

    optional[string]<\/p>\n<\/td>\n

    		\n

    \u043e\u0442\u0447\u0435\u0441\u0442\u0432\u043e \u043a\u043b\u0438\u0435\u043d\u0442\u0430 (\u043d\u0435\u043e\u0431\u044f\u0437\u0430\u0442\u0435\u043b\u044c\u043d\u043e\u0435 \u043f\u043e\u043b\u0435)<\/p>\n<\/td>\n<\/tr>\n

    \n

    sex<\/p>\n<\/td>\n

    		\n

    string<\/p>\n<\/td>\n

    		\n

    \u043f\u043e\u043b<\/p>\n<\/td>\n<\/tr>\n<\/tbody>\n<\/table>\n<\/div>\n<\/div>\n

      \n
    1. \n

      ClientCompany<\/em> — \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044f \u043e \u043a\u043e\u043c\u043f\u0430\u043d\u0438\u0438 \u043a\u043b\u0438\u0435\u043d\u0442\u0430:<\/p>\n<\/li>\n<\/ol>\n

      \n
      \n\n\n\n\n\n\n
      \n

      \u0418\u043c\u044f \u043f\u043e\u043b\u044f<\/p>\n<\/th>\n

      		\n

      \u0422\u0438\u043f \u043f\u043e\u043b\u044f<\/p>\n<\/th>\n

      		\n

      \u041e\u043f\u0438\u0441\u0430\u043d\u0438\u0435<\/p>\n<\/th>\n<\/tr>\n

      \n

      clientId<\/p>\n<\/td>\n

      		\n

      int<\/p>\n<\/td>\n

      		\n

      \u0443\u043d\u0438\u043a\u0430\u043b\u044c\u043d\u044b\u0439 \u0438\u0434\u0435\u043d\u0442\u0438\u0444\u0438\u043a\u0430\u0442\u043e\u0440 \u043a\u043b\u0438\u0435\u043d\u0442\u0430<\/p>\n<\/td>\n<\/tr>\n

      \n

      companyId<\/p>\n<\/td>\n

      		\n

      int<\/p>\n<\/td>\n

      		\n

      \u0443\u043d\u0438\u043a\u0430\u043b\u044c\u043d\u044b\u0439 \u0438\u0434\u0435\u043d\u0442\u0438\u0444\u0438\u043a\u0430\u0442\u043e\u0440 \u043a\u043e\u043c\u043f\u0430\u043d\u0438\u0438<\/p>\n<\/td>\n<\/tr>\n

      \n

      companyName<\/p>\n<\/td>\n

      		\n

      string<\/p>\n<\/td>\n

      		\n

      \u043d\u0430\u0437\u0432\u0430\u043d\u0438\u0435 \u043a\u043e\u043c\u043f\u0430\u043d\u0438\u0438<\/p>\n<\/td>\n<\/tr>\n<\/tbody>\n<\/table>\n<\/div>\n<\/div>\n

        \n
      1. \n

        Payment<\/em> — \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044f \u043e \u043f\u043b\u0430\u0442\u0435\u0436\u0430\u0445 \u043a\u043b\u0438\u0435\u043d\u0442\u0430:<\/p>\n<\/li>\n<\/ol>\n

        \n
        \n\n\n\n\n\n\n
        \n

        \u0418\u043c\u044f \u043f\u043e\u043b\u044f<\/p>\n<\/th>\n

        		\n

        \u0422\u0438\u043f \u043f\u043e\u043b\u044f<\/p>\n<\/th>\n

        		\n

        \u041e\u043f\u0438\u0441\u0430\u043d\u0438\u0435<\/p>\n<\/th>\n<\/tr>\n

        \n

        clientId<\/p>\n<\/td>\n

        		\n

        int<\/p>\n<\/td>\n

        		\n

        \u0443\u043d\u0438\u043a\u0430\u043b\u044c\u043d\u044b\u0439 \u0438\u0434\u0435\u043d\u0442\u0438\u0444\u0438\u043a\u0430\u0442\u043e\u0440 \u043a\u043b\u0438\u0435\u043d\u0442\u0430<\/p>\n<\/td>\n<\/tr>\n

        \n

        amount<\/p>\n<\/td>\n

        		\n

        int<\/p>\n<\/td>\n

        		\n

        \u0441\u0443\u043c\u043c\u0430 \u043f\u043b\u0430\u0442\u0435\u0436\u0430<\/p>\n<\/td>\n<\/tr>\n

        \n

        tmMs<\/p>\n<\/td>\n

        		\n

        long<\/p>\n<\/td>\n

        		\n

        \u0432\u0440\u0435\u043c\u044f \u0441\u0432\u0435\u0440\u0448\u0435\u043d\u0438\u044f \u043f\u043b\u0430\u0442\u0435\u0436\u0430<\/p>\n<\/td>\n<\/tr>\n<\/tbody>\n<\/table>\n<\/div>\n<\/div>\n

        \u0411\u0443\u0434\u0435\u043c \u043f\u043e\u043b\u0443\u0447\u0430\u0442\u044c \u043e\u0431\u043d\u043e\u0432\u043b\u0435\u043d\u0438\u044f \u043f\u043e \u0432\u0441\u0435\u043c \u0442\u0430\u0431\u043b\u0438\u0446\u0430\u043c \u0447\u0435\u0440\u0435\u0437 CDC<\/a>-\u043a\u0430\u043d\u0430\u043b (\u043f\u0443\u0441\u0442\u044c \u044d\u0442\u043e \u0431\u0443\u0434\u0435\u0442 Debezium<\/a>) \u0432 \u0442\u043e\u043f\u0438\u043a Kafka<\/a> \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 Json<\/a>. \u0417\u0430\u0434\u0430\u0447\u0435\u0439 \u043d\u0430\u0448\u0435\u0439 Job<\/em>-\u044b \u0431\u0443\u0434\u0435\u0442 \u0441\u043b\u0443\u0448\u0430\u0442\u044c \u0442\u043e\u043f\u0438\u043a\u0438 \u0438 \u0432\u044b\u0434\u0430\u0432\u0430\u0442\u044c \u043e\u0431\u043d\u043e\u0432\u043b\u0435\u043d\u0438\u044f \u043f\u043e \u043a\u043b\u0438\u0435\u043d\u0442\u0443 (\u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440: \u043a\u043b\u0438\u0435\u043d\u0442 \u0441\u043c\u0435\u043d\u0438\u043b \u043a\u043e\u043c\u043f\u0430\u043d\u0438\u044e) \u0432\u043e \u0432\u043d\u0435\u0448\u043d\u044e\u044e \u0441\u0438\u0441\u0442\u0435\u043c\u0443 (\u0432 \u0440\u0430\u043c\u043a\u0430\u0445 \u0441\u0442\u0430\u0442\u044c\u0438 \u0432 \u043a\u0430\u0447\u0435\u0441\u0442\u0432\u0435 \u0432\u043d\u0435\u0448\u043d\u0435\u0439 \u0441\u0438\u0441\u0442\u0435\u043c\u044b \u0431\u0443\u0434\u0435\u043c \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c \u043a\u043e\u043d\u0441\u043e\u043b\u044c \u0432\u044b\u0432\u043e\u0434\u0430).<\/p>\n

        \"\u0418\u0437\u043e\u0431\u0440\u0430\u0436\u0435\u043d\u0438\u0435\"<\/p>\n
        \u0420\u0438\u0441\u0443\u043d\u043e\u043a 1 — \u0421\u0445\u0435\u043c\u0430 \u0431\u0438\u0437\u043d\u0435\u0441\u0441 \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u0430<\/figcaption><\/div>\n<\/figure>\n

        \u041d\u0430\u0447\u0438\u043d\u0430\u0435\u043c \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0438\u0440\u043e\u0432\u0430\u0442\u044c<\/h2>\n

        \u041d\u0430\u0448 \u0442\u0435\u0445\u043d\u043e\u043b\u043e\u0433\u0438\u0447\u0435\u0441\u043a\u0438\u0439 \u0441\u0442\u0435\u043a:<\/p>\n

          \n
        1. \n

          Scala<\/a> v2.12.10<\/p>\n<\/li>\n

        2. \n

          Flink<\/a> v1.16.0<\/p>\n<\/li>\n

        3. \n

          Circe<\/a> v0.14.3<\/p>\n<\/li>\n<\/ol>\n

          \u041c\u044b \u0440\u0430\u0437\u0431\u0435\u0440\u0435\u043c 2 \u0432\u0430\u0440\u0438\u0430\u043d\u0442\u0430 \u0440\u0435\u0448\u0435\u043d\u0438\u044f:<\/p>\n

            \n
          1. \n

            \u041f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 Table API<\/a><\/p>\n<\/li>\n

          2. \n

            \u041f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 \u043a\u043b\u0430\u0441\u0441\u0438\u0447\u0435\u0441\u043a\u043e\u0433\u043e State Processor API<\/a><\/p>\n<\/li>\n<\/ol>\n

            \u041f\u0440\u0435\u0436\u0434\u0435 \u0447\u0435\u043c \u043f\u0438\u0441\u0430\u0442\u044c \u043e\u0441\u043d\u043e\u0432\u043d\u0443\u044e \u043b\u043e\u0433\u0438\u043a\u0443 \u043d\u0430\u043c \u043f\u043e\u0442\u0440\u0435\u0431\u0443\u0435\u0442\u0441\u044f \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u043e\u0431\u0449\u0438\u0445 \u0448\u0430\u0433\u043e\u0432, \u043a\u043e\u0442\u043e\u0440\u044b\u043c\u0438 \u043c\u044b \u0431\u0443\u0434\u0435\u043c \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c\u0441\u044f \u0432 \u043e\u0431\u0435\u0438\u0445 \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044f\u0445.<\/p>\n

            \u041f\u0435\u0440\u0432\u043e\u0435 \u0447\u0442\u043e \u043d\u0430\u043c \u043f\u043e\u0442\u0440\u0435\u0431\u0443\u0435\u0442\u0441\u044f — \u044d\u0442\u043e \u043d\u0430\u0443\u0447\u0438\u0442\u044c\u0441\u044f \u0447\u0438\u0442\u0430\u0442\u044c \u0442\u043e\u043f\u0438\u043a\u0438 Kafka<\/em>, \u0432 \u044d\u0442\u043e\u043c \u043d\u0430\u043c \u043f\u043e\u043c\u043e\u0436\u0435\u0442 \u0443\u0436\u0435 \u0433\u043e\u0442\u043e\u0432\u044b\u0439 Flink Kafka Connector<\/a>:<\/p>\n

            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 \/\/ \u0432 \u0441\u043b\u0443\u0447\u0430\u0435 \u043d\u0435\u0443\u0434\u0430\u0447\u0438 \u043f\u0440\u0438 \u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438, \u043e\u0448\u0438\u0431\u043a\u0430 \u043f\u0440\u043e\u0441\u0442\u043e \u0438\u0433\u043d\u043e\u0440\u0438\u0440\u0443\u0435\u0442\u0441\u044f \u0438 \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u043e\u0442\u0431\u0440\u0430\u0441\u044b\u0432\u0430\u0435\u0442\u0441\u044f (\u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u044b\u0445 \u043f\u0440\u043e\u0435\u043a\u0442\u0430\u0445 \u0442\u0430\u043a \u0434\u0435\u043b\u0430\u0442\u044c \u043d\u0435 \u0441\u0442\u043e\u0438\u0442)     obj.foreach(out.collect)   }    override def getProducedType: TypeInformation[A] = {     TypeExtractor       .getForClass(classTag[A].runtimeClass.asInstanceOf[Class[A]])       .asInstanceOf[TypeInformation[A]]   } } <\/code><\/pre>\n
            \u0412 \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438\/\u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 \u043d\u0430\u043c \u043f\u043e\u043c\u043e\u0436\u0435\u0442 Circe<\/em>:<\/p>\n
            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() } <\/code><\/pre>\n
            \u041c\u044b \u043d\u0435\u043c\u043d\u043e\u0433\u043e \u0443\u043f\u0440\u043e\u0441\u0442\u0438\u043c \u0441\u0435\u0431\u0435 \u0437\u0430\u0434\u0430\u0447\u0443, \u0441\u044b\u043c\u0438\u0442\u0438\u0440\u0443\u0435\u043c \u043f\u043e\u0432\u0435\u0434\u0435\u043d\u0438\u0435 CDC<\/em>-\u043a\u0430\u043d\u0430\u043b\u0430, \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u044f \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0442\u0440\u0435\u0431\u0443\u0435\u043c\u043e\u0433\u043e \u0444\u043e\u0440\u043c\u0430\u0442\u0430 \u043d\u0430\u043f\u0440\u044f\u043c\u0443\u044e \u0432 \u0442\u043e\u043f\u0438\u043a Kafka<\/em>:<\/p>\n
            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) }  <\/code><\/pre>\n
            \u0422\u0435\u043f\u0435\u0440\u044c \u043c\u044b \u0433\u043e\u0442\u043e\u0432\u044b \u043f\u0440\u0438\u0441\u0442\u0443\u043f\u0438\u0442\u044c \u043a \u043e\u0441\u043d\u043e\u0432\u043d\u043e\u0439 \u0447\u0430\u0441\u0442\u0438.<\/p>\n
            \u0418\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u043d\u0438\u0435 Flink Table API<\/h2>\n
            \u0418\u0442\u0430\u043a, \u043f\u0440\u0435\u0436\u0434\u0435 \u0432\u0441\u0435\u0433\u043e \u043d\u0430\u043c \u043d\u0443\u0436\u043d\u043e \u043f\u043e\u043b\u0443\u0447\u0438\u0442\u044c \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e\u0435 \u043e\u043a\u0440\u0443\u0436\u0435\u043d\u0438\u0435, \u043f\u043e\u0441\u043a\u043e\u043b\u044c\u043a\u0443 \u043c\u044b \u0431\u0443\u0434\u0435\u043c \u0440\u0430\u0431\u043e\u0442\u0430\u0442\u044c \u0432 \u043f\u043e\u0442\u043e\u043a\u043e\u0432\u043e\u043c \u0440\u0435\u0436\u0438\u043c\u0435, \u0437\u0430\u0440\u0443\u0447\u0438\u043c\u0441\u044f \u0442\u0430\u043a\u043e\u0432\u044b\u043c:<\/p>\n
            implicit val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment <\/code><\/pre>\n
            \u0414\u0430\u043b\u0435\u0435 \u043f\u043e\u043b\u0443\u0447\u0438\u043c \u043e\u043a\u0440\u0443\u0436\u0435\u043d\u0438\u0435 \u0434\u043b\u044f \u0434\u043e\u0441\u0442\u0443\u043f\u0430 \u043a Table API<\/em> :<\/p>\n
            implicit val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env) <\/code><\/pre>\n
            \u0424\u043e\u0440\u043c\u0438\u0440\u0443\u0435\u043c Kafka Source<\/em> \u043f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 \u043a\u043e\u0434\u0430<\/a> \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u043c\u044b \u043f\u043e\u0434\u0433\u043e\u0442\u043e\u0432\u0438\u043b\u0438 \u0432\u044b\u0448\u0435:<\/p>\n
            val clients: DataStream[Client] = KafkaConsumerSource.configureKafkaDataSource[Client](\"clients\") <\/code><\/pre>\n
            \u0422\u0435\u043f\u0435\u0440\u044c \u043d\u0430\u043c \u043e\u0441\u0442\u0430\u043b\u0441\u044f \u0432\u0441\u0435\u0433\u043e \u043e\u0434\u0438\u043d \u0448\u0430\u0433 \u0434\u043e \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u044f \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u0435\u043d\u0438\u044f \u0441 \u043a\u043e\u0442\u043e\u0440\u044b\u043c \u043d\u0430\u043c \u043f\u0440\u0435\u0434\u0441\u0442\u043e\u0438\u0442 \u0440\u0430\u0431\u043e\u0442\u0430\u0442\u044c — Dynamic Tables<\/a>.  \u0427\u0442\u043e\u0431\u044b \u0441\u0434\u0435\u043b\u0430\u0442\u044c \u044d\u0442\u043e\u0442 \u0448\u0430\u0433 \u043c\u0430\u043a\u0441\u0438\u043c\u0430\u043b\u044c\u043d\u043e \u043f\u0440\u043e\u0441\u0442\u044b\u043c, \u0432\u043e\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u043c\u0441\u044f \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044f\u043c\u0438 Scala<\/a> \u0438 \u0440\u0430\u0441\u0448\u0438\u0440\u0438\u043c DataStream<\/em> \u043c\u0435\u0442\u043e\u0434\u043e\u043c toStreamTable<\/em>:<\/p>\n
            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()       )   } } <\/code><\/pre>\n
            \u041d\u0430 \u0432\u0445\u043e\u0434 \u0444\u0443\u043d\u043a\u0446\u0438\u0438 \u043f\u043e\u0434\u0430\u0435\u0442\u0441\u044f \u0441\u043f\u0438\u0441\u043e\u043a \u0438\u043c\u0435\u043d \u043f\u043e\u043b\u0435\u0439, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u0441\u043e\u0441\u0442\u0430\u0432\u044f\u0442 \u043f\u0435\u0440\u0432\u0438\u0447\u043d\u044b\u0439 \u043a\u043b\u044e\u0447 \u043d\u0430\u0448\u0435\u0439 \u0442\u0430\u0431\u043b\u0438\u0446\u044b. \u041a\u0430\u043a \u0438 \u0432 \u0431\u043e\u043b\u044c\u0448\u0438\u043d\u0441\u0442\u0432\u0435 \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a\u043e\u0432 \u043d\u0430\u0438\u043c\u0435\u043d\u044c\u0448\u0435\u0439 \u0435\u0434\u0438\u043d\u0438\u0446\u0435\u0439 \u0434\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b \u0441 \u0442\u0430\u0431\u043b\u0438\u0446\u0430\u043c\u0438 \u044f\u0432\u043b\u044f\u0435\u0442\u0441\u044f \u0441\u0442\u0440\u043e\u043a\u0430 (Row<\/a>). \u041d\u0443\u0436\u0435\u043d \u0441\u043f\u043e\u0441\u043e\u0431 \u043f\u0440\u0435\u043e\u0431\u0440\u0430\u0437\u043e\u0432\u0430\u043d\u0438\u044f \u0432\u0441\u0435\u0445 \u043d\u0430\u0448\u0438\u0445 \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0435\u0439 \u0432 \u0442\u0438\u043f Row<\/em>, \u043f\u0440\u043e\u0434\u043e\u043b\u0436\u0438\u043c \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c\u0441\u044f \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044f\u043c\u0438 Scala<\/em>, \u043d\u0430 \u044d\u0442\u043e\u0442 \u0440\u0430\u0437 \u0442\u0435\u043c \u0447\u0442\u043e Scala<\/em> \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u0435\u0442 \u043f\u0438\u0441\u0430\u0442\u044c \u043a\u043e\u0434 \u0432 \u0444\u0443\u043d\u043a\u0446\u0438\u043e\u043d\u0430\u043b\u044c\u043d\u043e\u043c \u0441\u0442\u0438\u043b\u0435<\/a>, \u0430 \u043a\u043e\u043d\u043a\u0440\u0435\u0442\u043d\u043e \u043c\u044b \u0432\u043e\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u043c\u0441\u044f Type Class<\/a>-\u043c\u0438. \u041d\u0430\u0448 Type Class<\/em> \u0431\u0443\u0434\u0435\u0442 \u0438\u043c\u0435\u0442\u044c \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u0439 \u0432\u0438\u0434:<\/p>\n
            sealed trait RowTransformer[A] {   def toRow(entity: A): Option[Row]   protected val fieldsName: Array[String]   implicit def typeInformation: TypeInformation[Row] } <\/code><\/pre>\n
            \u041f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u0438\u043c \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044e \u0434\u043b\u044f \u043e\u0434\u043d\u043e\u0439 \u0438\u0437 \u0442\u0430\u0431\u043b\u0438\u0446 (\u0434\u043b\u044f \u043e\u0441\u0442\u0430\u043b\u044c\u043d\u044b\u0445 \u0430\u043b\u0433\u043e\u0440\u0438\u0442\u043c \u0431\u0443\u0434\u0435\u0442 \u0430\u043d\u0430\u043b\u043e\u0433\u0438\u0447\u0435\u043d):<\/p>\n
            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\") } <\/code><\/pre>\n
            \u0422\u0435\u043f\u0435\u0440\u044c \u043c\u044b \u043c\u043e\u0436\u0435\u043c \u043f\u0440\u0435\u043e\u0431\u0440\u0430\u0437\u043e\u0432\u044b\u0432\u0430\u0442\u044c DataStream<\/code> \u0432 Table<\/code> \u043e\u0434\u043d\u0438\u043c \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u0435\u043c:<\/p>\n
            val clients: Table = KafkaConsumerSource.configureKafkaDataSource[Client](\"clients\")   .toStreamTable(\"clientId\") <\/code><\/pre>\n
            \u042d\u0442\u043e \u0434\u0430\u043b\u0435\u043a\u043e \u043d\u0435 \u0435\u0434\u0438\u043d\u0441\u0442\u0432\u0435\u043d\u043d\u044b\u0439 \u0441\u043f\u043e\u0441\u043e\u0431 \u0441\u043e\u0437\u0434\u0430\u043d\u0438\u044f \u0442\u0430\u0431\u043b\u0438\u0446, \u0435\u0441\u043b\u0438 \u043c\u044b \u0438\u0437\u043d\u0430\u0447\u0430\u043b\u044c\u043d\u043e \u0445\u043e\u0442\u0438\u043c \u0440\u0430\u0431\u043e\u0442\u0430\u0442\u044c \u0441 \u0434\u0430\u043d\u043d\u044b\u043c\u0438 \u043a\u0430\u043a \u0441 \u0442\u0430\u0431\u043b\u0438\u0446\u0435\u0439, \u043d\u0435\u043e\u0431\u044f\u0437\u0430\u0442\u0435\u043b\u044c\u043d\u043e \u0441\u043f\u0435\u0440\u0432\u0430 \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u044f\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 \u0432 \u0432\u0438\u0434\u0435 DataStream<\/em>:<\/p>\n
            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\") <\/code><\/pre>\n
            \u0418 \u043a\u043e\u043d\u0435\u0447\u043d\u043e \u0436\u0435 \u043a\u0443\u0434\u0430 \u0431\u0435\u0437 SQL<\/em>:<\/p>\n
            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\") <\/code><\/pre>\n
            \u041c\u044b \u043e\u043a\u043e\u043d\u0447\u0430\u0442\u0435\u043b\u044c\u043d\u043e \u043f\u043e\u0434\u0433\u043e\u0442\u043e\u0432\u0438\u043b\u0438\u0441\u044c \u043a \u043d\u0430\u043f\u0438\u0441\u0430\u043d\u0438\u044e \u043e\u0441\u043d\u043e\u0432\u043d\u043e\u0439 \u043b\u043e\u0433\u0438\u043a\u0438. \u0414\u043b\u044f \u043d\u0430\u0448\u0435\u0439 \u0437\u0430\u0434\u0430\u0447\u0438 \u0431\u0443\u0434\u0435\u043c \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c inner join<\/em>, \u0442.\u0435. \u043f\u0440\u0438 \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u0438 \u043d\u043e\u0432\u043e\u0433\u043e \u0441\u043e\u0431\u044b\u0442\u0438\u044f \u043c\u044b \u0431\u0443\u0434\u0435\u043c \u0432\u044b\u0434\u0430\u0432\u0430\u0442\u044c \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442 \u0432\u043e \u0432\u043d\u0435\u0448\u043d\u044e\u044e \u0441\u0438\u0441\u0442\u0435\u043c\u0443<\/em> \u0442\u043e\u043b\u044c\u043a\u043e \u0432 \u0441\u043b\u0443\u0447\u0430\u0435 \u043d\u0430\u043b\u0438\u0447\u0438\u044f \u0434\u0430\u043d\u043d\u044b\u0445 \u043f\u043e \u043a\u043b\u044e\u0447\u0443 \u0432\u043e \u0432\u0441\u0435\u0445 \u0442\u0430\u0431\u043b\u0438\u0446\u0430\u0445.<\/p>\n
            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\") <\/code><\/pre>\n
            \"\u0418\u0437\u043e\u0431\u0440\u0430\u0436\u0435\u043d\u0438\u0435\"<\/p>\n
            \u0420\u0438\u0441\u0443\u043d\u043e\u043a 2 — Dynamic Tables<\/figcaption><\/div>\n<\/figure>\n
            \u0418\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u043d\u0438\u0435 State Processor API<\/h2>\n
            \u0421\u043d\u043e\u0432\u0430 \u043d\u0430\u0447\u043d\u0435\u043c \u0441 \u043e\u043a\u0440\u0443\u0436\u0435\u043d\u0438\u044f, \u0432 \u044d\u0442\u043e\u043c \u0432\u0430\u0440\u0438\u0430\u043d\u0442\u0435 \u043d\u0430\u043c \u0431\u0443\u0434\u0435\u0442 \u0434\u043e\u0441\u0442\u0430\u0442\u043e\u0447\u043d\u043e \u0442\u043e\u043b\u044c\u043a\u043e \u043e\u043a\u0440\u0443\u0436\u0435\u043d\u0438\u044f \u0434\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b \u0432 \u043f\u043e\u0442\u043e\u043a\u043e\u0432\u043e\u043c \u0440\u0435\u0436\u0438\u043c\u0435:<\/p>\n
            implicit val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment <\/code><\/pre>\n
            \u0412 \u043f\u0440\u043e\u0448\u043b\u043e\u043c \u0432\u0430\u0440\u0438\u0430\u043d\u0442\u0435 \u043c\u044b \u043f\u0440\u0438\u0432\u043e\u0434\u0438\u043b\u0438 \u0432\u0441\u0435 \u043d\u0430\u0448\u0438 \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0438 \u043a \u0442\u0438\u043f\u0443 Table<\/em>, \u0441\u0435\u0439\u0447\u0430\u0441 \u043d\u0430\u043c \u043d\u0443\u0436\u043d\u043e \u0441\u0434\u0435\u043b\u0430\u0442\u044c \u043d\u0435\u0447\u0442\u043e \u043f\u043e\u0445\u043e\u0436\u0435\u0435:<\/p>\n
            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       )     }   } } <\/code><\/pre>\n
            \u041c\u044b \u043e\u043f\u0438\u0441\u0430\u043b\u0438 \u0441\u0432\u043e\u0439 \u0441\u043e\u0431\u0441\u0442\u0432\u0435\u043d\u043d\u044b\u0439 \u0430\u043d\u0430\u043b\u043e\u0433 \u0442\u0430\u0431\u043b\u0438\u0446\u044b, \u0434\u043e\u0431\u0430\u0432\u0438\u0432 \u0432 \u043d\u0435\u0433\u043e \u043c\u0435\u0442\u043e\u0434 complete<\/em> \u0444\u043e\u0440\u043c\u0438\u0440\u0443\u044e\u0449\u0438\u0439 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442 \u0442\u043e\u043b\u044c\u043a\u043e \u0432 \u0441\u043b\u0443\u0447\u0430\u0435 \u043d\u0430\u043b\u0438\u0447\u0438\u044f \u0434\u0430\u043d\u043d\u044b\u0445 \u043f\u043e \u0432\u0441\u0435\u043c \u0442\u0440\u0435\u043c \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u044f\u043c (\u0430\u043d\u0430\u043b\u043e\u0433 inner join<\/em>). \u041e\u043f\u0438\u0448\u0435\u043c \u0441\u043f\u043e\u0441\u043e\u0431 \u0442\u0440\u0430\u043d\u0441\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u0438 \u0438\u0441\u0445\u043e\u0434\u043d\u043e\u0439 \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0438 \u0432 \u043d\u0430\u0448 \u0442\u0438\u043f (\u0434\u043b\u044f \u043e\u0441\u0442\u0430\u043b\u044c\u043d\u044b\u0445 \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0435\u0439 \u0431\u0443\u0434\u0435\u0442 \u0430\u043d\u0430\u043b\u043e\u0433\u0438\u0447\u043d\u043e):<\/p>\n
            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)   } } <\/code><\/pre>\n
            \u0418\u0442\u043e\u0433\u043e\u0432\u044b\u0439 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442:<\/p>\n
            val clients: DataStream[PortfolioState] = KafkaConsumerSource.configureKafkaDataSource[Client](\"clients\").map(PortfolioState(_)) <\/code><\/pre>\n
            \u041f\u043e\u0441\u043b\u0435\u0434\u043d\u0435\u0435 \u0447\u0435\u0433\u043e \u043d\u0430\u043c \u043d\u0435 \u0445\u0432\u0430\u0442\u0430\u0435\u0442 — \u044d\u0442\u043e \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u0447\u0438\u043a\u0430<\/a> \u0441\u043e\u0431\u044b\u0442\u0438\u0439, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0431\u0443\u0434\u0435\u0442 \u043f\u043e\u0434\u0434\u0435\u0440\u0436\u0438\u0432\u0430\u0442\u044c \u0430\u043a\u0442\u0443\u0430\u043b\u044c\u043d\u043e\u0435 \u0441\u043e\u0441\u0442\u043e\u044f\u043d\u0438\u0435 \u0438 \u0432\u044b\u0434\u0430\u0432\u0430\u0442\u044c \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442\u044b:<\/p>\n
            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) \/\/ \u0435\u0441\u043b\u0438 state == null, \u0438\u043d\u0438\u0446\u0438\u0430\u043b\u0438\u0437\u0438\u0440\u0443\u0435\u043c \u0435\u0433\u043e \u0432\u0445\u043e\u0434\u043d\u044b\u043c \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435\u043c     } { 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) \/\/ \u0435\u0441\u043b\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u0445\u0432\u0430\u0442\u0430\u0435\u0442, \u0442\u043e \u0432\u044b\u0434\u0430\u0435\u043c \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442       state.update(updatedState)     }   } } <\/code><\/pre>\n
            \u0421\u043e\u0431\u0435\u0440\u0435\u043c \u0432\u0441\u0435 \u0432\u043c\u0435\u0441\u0442\u0435:<\/p>\n
            clients   .union(companies) \/\/ \u043e\u0431\u044a\u0435\u0434\u0438\u043d\u044f\u0435\u043c \u0432\u0441\u0435 \u0442\u0440\u0438 \u043f\u043e\u0442\u043e\u043a\u0430 \u0432 \u043e\u0434\u0438\u043d   .union(payments)   .keyBy(_.id) \/\/ \u0448\u0430\u0440\u0434\u0438\u0443\u0440\u0435\u043c \u0434\u0430\u043d\u043d\u044b\u0435 \u043f\u043e \u043f\u043e\u043b\u044e id (clientId)   .process(new PortfolioStateProcessor())   .print(\"Portfolio\") <\/code><\/pre>\n
            \u0418\u0442\u043e\u0433\u0438<\/h2>\n
            \u041c\u044b \u0440\u0430\u0441\u0441\u043c\u043e\u0442\u0440\u0435\u043b\u0438 2 \u0440\u0430\u0437\u043d\u044b\u0445 \u043f\u043e\u0434\u0445\u043e\u0434\u0430, \u0434\u0430\u0436\u0435 \u043d\u0435\u0441\u043c\u043e\u0442\u0440\u044f \u043d\u0430 \u0432\u0441\u044e \u043f\u0440\u043e\u0441\u0442\u043e\u0442\u0443 \u043f\u043e\u0441\u0442\u0430\u043d\u043e\u0432\u043a\u0438 \u043f\u043e\u0434\u0445\u043e\u0434 \u0441 \u043f\u0440\u0438\u043c\u0435\u043d\u0435\u043d\u0438\u0435\u043c Table API<\/em> \u0434\u043b\u044f \u0437\u0430\u0434\u0430\u0447 \u043e\u0442\u0441\u043b\u0435\u0436\u0438\u0432\u0430\u043d\u0438\u044f \u0438\u0437\u043c\u0435\u043d\u0435\u043d\u0438\u0439 \u0438 \u043c\u043e\u043c\u0435\u043d\u0442\u0430\u043b\u044c\u043d\u043e\u0439 \u0432\u044b\u0434\u0430\u0447\u0438 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442 \u0432\u044b\u0433\u043b\u044f\u0434\u0438\u0442 \u043f\u0440\u0438\u0432\u043b\u0435\u043a\u0430\u0442\u0435\u043b\u044c\u043d\u0435\u0435, \u043c\u044b \u043f\u043e\u043b\u0443\u0447\u0430\u0435\u043c \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044c \u043e\u043f\u0438\u0441\u044b\u0432\u0430\u0442\u044c \u0436\u0435\u043b\u0430\u0435\u043c\u044b\u0439 \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442 \u043d\u0430 \u043f\u0440\u0438\u0432\u044b\u0447\u043d\u043e\u043c SQL<\/em>-\u043f\u043e\u0434\u043e\u0431\u043d\u043e\u043c \u0441\u0438\u043d\u0442\u0430\u043a\u0441\u0438\u0441\u0435 \u043d\u0430\u0445\u043e\u0434\u044f\u0441\u044c \u0432 \u0431\u0435\u0441\u043a\u043e\u043d\u0435\u0447\u043d\u043e\u043c \u043f\u043e\u0442\u043e\u043a\u0435 \u0441\u043e\u0431\u044b\u0442\u0438\u0439.<\/p>\n
            \u041f\u043e\u043b\u043d\u044b\u0439 \u043a\u043e\u0434 \u043f\u0440\u043e\u0435\u043a\u0442\u0430 \u043c\u043e\u0436\u043d\u043e \u043d\u0430\u0439\u0442\u0438 \u0437\u0434\u0435\u0441\u044c<\/a><\/p>\n<\/div>\n<\/div>\n<\/div>\n
              <\/div>\n
              
             \u0441\u0441\u044b\u043b\u043a\u0430 \u043d\u0430 \u043e\u0440\u0438\u0433\u0438\u043d\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438              https:\/\/habr.com\/ru\/companies\/neoflex\/articles\/745730\/<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"
            <\/div>\n
            \n
            \n
            \n
            \n
            <\/figure>\n
            \u0412\u0441\u0435\u043c \u043f\u0440\u0438\u0432\u0435\u0442!<\/p>\n
            \u0412 \u044d\u0442\u043e\u0439 \u0441\u0442\u0430\u0442\u044c\u0435 \u043c\u044b \u043f\u043e\u0441\u043c\u043e\u0442\u0440\u0438\u043c \u043d\u0430 \u0434\u0432\u0430 \u0432\u0430\u0440\u0438\u0430\u043d\u0442\u0430 \u0440\u0435\u0448\u0435\u043d\u0438\u044f «\u0442\u0438\u043f\u0438\u0447\u043d\u043e\u0439» \u0437\u0430\u0434\u0430\u0447\u0438 \u043f\u043e\u0442\u043e\u043a\u043e\u0432\u043e\u0439 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 \u0441 \u0443\u0447\u0435\u0442\u043e\u043c \u0441\u043e\u0441\u0442\u043e\u044f\u043d\u0438\u044f \u043f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a\u0430 Apache Flink<\/em>.<\/p>\n
            \u0415\u0441\u043b\u0438 \u0432\u044b \u0442\u043e\u043b\u044c\u043a\u043e \u043d\u0430\u0447\u0438\u043d\u0430\u0435\u0442\u0435 \u0437\u043d\u0430\u043a\u043e\u043c\u0441\u0442\u0432\u043e \u0441 Flink<\/em>, \u0437\u0434\u0435\u0441\u044c<\/a> \u043c\u043e\u0436\u043d\u043e \u043d\u0430\u0439\u0442\u0438 \u043e\u0431\u0438\u043b\u044c\u043d\u043e\u0435 \u043a\u043e\u043b\u0438\u0447\u0435\u0441\u0442\u0432\u043e \u043c\u0430\u0442\u0435\u0440\u0438\u0430\u043b\u043e\u0432.<\/p>\n
            \u041f\u043e\u0441\u0442\u0430\u043d\u043e\u0432\u043a\u0430 \u0437\u0430\u0434\u0430\u0447\u0438<\/h2>\n
            \u0418\u0442\u0430\u043a, \u0434\u0430\u0432\u0430\u0439\u0442\u0435 \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u0438\u043c \u0431\u0430\u0437\u0443 \u0434\u0430\u043d\u043d\u044b\u0445 (\u043f\u0443\u0441\u0442\u044c \u044d\u0442\u043e \u0431\u0443\u0434\u0435\u0442 PostgreSQL<\/em>) \u0441\u043e\u0441\u0442\u043e\u044f\u0449\u0443\u044e \u0438\u0437 \u0442\u0440\u0435\u0445 \u0442\u0430\u0431\u043b\u0438\u0446:<\/p>\n
              \n
            1. \n
              Clients<\/em> — \u043e\u0431\u0449\u0430\u044f \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044f \u043f\u043e \u043a\u043b\u0438\u0435\u043d\u0442\u0443:<\/p>\n<\/li>\n<\/ol>\n
              \n
              \n\n\n\n\n\n\n\n\n
              \n
              \u0418\u043c\u044f \u043f\u043e\u043b\u044f<\/p>\n<\/th>\n
              		\n
              \u0422\u0438\u043f \u043f\u043e\u043b\u044f<\/p>\n<\/th>\n
              		\n
              \u041e\u043f\u0438\u0441\u0430\u043d\u0438\u0435<\/p>\n<\/th>\n<\/tr>\n
              \n
              clientId<\/p>\n<\/td>\n
              		\n
              int<\/p>\n<\/td>\n
              		\n
              \u0443\u043d\u0438\u043a\u0430\u043b\u044c\u043d\u044b\u0439 \u0438\u0434\u0435\u043d\u0442\u0438\u0444\u0438\u043a\u0430\u0442\u043e\u0440 \u043a\u043b\u0438\u0435\u043d\u0442\u0430<\/p>\n<\/td>\n<\/tr>\n
              \n
              name<\/p>\n<\/td>\n
              		\n
              string<\/p>\n<\/td>\n
              		\n
              \u0438\u043c\u044f \u043a\u043b\u0438\u0435\u043d\u0442\u0430<\/p>\n<\/td>\n<\/tr>\n
              \n
              surname<\/p>\n<\/td>\n
              		\n
              string<\/p>\n<\/td>\n
              		\n
              \u0444\u0430\u043c\u0438\u043b\u0438\u044f \u043a\u043b\u0438\u0435\u043d\u0442\u0430<\/p>\n<\/td>\n<\/tr>\n
              \n
              patronymic<\/p>\n<\/td>\n
              		\n
              optional[string]<\/p>\n<\/td>\n
              		\n
              \u043e\u0442\u0447\u0435\u0441\u0442\u0432\u043e \u043a\u043b\u0438\u0435\u043d\u0442\u0430 (\u043d\u0435\u043e\u0431\u044f\u0437\u0430\u0442\u0435\u043b\u044c\u043d\u043e\u0435 \u043f\u043e\u043b\u0435)<\/p>\n<\/td>\n<\/tr>\n
              \n
              sex<\/p>\n<\/td>\n
              		\n
              string<\/p>\n<\/td>\n
              		\n
              \u043f\u043e\u043b<\/p>\n<\/td>\n<\/tr>\n<\/tbody>\n<\/table>\n<\/div>\n<\/div>\n
                \n
              1. \n
                ClientCompany<\/em> — \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044f \u043e \u043a\u043e\u043c\u043f\u0430\u043d\u0438\u0438 \u043a\u043b\u0438\u0435\u043d\u0442\u0430:<\/p>\n<\/li>\n<\/ol>\n
                \n
                \n\n\n\n\n\n\n
                \n
                \u0418\u043c\u044f \u043f\u043e\u043b\u044f<\/p>\n<\/th>\n
                		\n
                \u0422\u0438\u043f \u043f\u043e\u043b\u044f<\/p>\n<\/th>\n
                		\n
                \u041e\u043f\u0438\u0441\u0430\u043d\u0438\u0435<\/p>\n<\/th>\n<\/tr>\n
                \n
                clientId<\/p>\n<\/td>\n
                		\n
                int<\/p>\n<\/td>\n
                		\n
                \u0443\u043d\u0438\u043a\u0430\u043b\u044c\u043d\u044b\u0439 \u0438\u0434\u0435\u043d\u0442\u0438\u0444\u0438\u043a\u0430\u0442\u043e\u0440 \u043a\u043b\u0438\u0435\u043d\u0442\u0430<\/p>\n<\/td>\n<\/tr>\n
                \n
                companyId<\/p>\n<\/td>\n
                		\n
                int<\/p>\n<\/td>\n
                		\n
                \u0443\u043d\u0438\u043a\u0430\u043b\u044c\u043d\u044b\u0439 \u0438\u0434\u0435\u043d\u0442\u0438\u0444\u0438\u043a\u0430\u0442\u043e\u0440 \u043a\u043e\u043c\u043f\u0430\u043d\u0438\u0438<\/p>\n<\/td>\n<\/tr>\n
                \n
                companyName<\/p>\n<\/td>\n
                		\n
                string<\/p>\n<\/td>\n
                		\n
                \u043d\u0430\u0437\u0432\u0430\u043d\u0438\u0435 \u043a\u043e\u043c\u043f\u0430\u043d\u0438\u0438<\/p>\n<\/td>\n<\/tr>\n<\/tbody>\n<\/table>\n<\/div>\n<\/div>\n
                  \n
                1. \n
                  Payment<\/em> — \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044f \u043e \u043f\u043b\u0430\u0442\u0435\u0436\u0430\u0445 \u043a\u043b\u0438\u0435\u043d\u0442\u0430:<\/p>\n<\/li>\n<\/ol>\n
                  \n
                  \n\n\n\n\n\n\n
                  \n
                  \u0418\u043c\u044f \u043f\u043e\u043b\u044f<\/p>\n<\/th>\n
                  		\n
                  \u0422\u0438\u043f \u043f\u043e\u043b\u044f<\/p>\n<\/th>\n
                  		\n
                  \u041e\u043f\u0438\u0441\u0430\u043d\u0438\u0435<\/p>\n<\/th>\n<\/tr>\n
                  \n
                  clientId<\/p>\n<\/td>\n
                  		\n
                  int<\/p>\n<\/td>\n
                  		\n
                  \u0443\u043d\u0438\u043a\u0430\u043b\u044c\u043d\u044b\u0439 \u0438\u0434\u0435\u043d\u0442\u0438\u0444\u0438\u043a\u0430\u0442\u043e\u0440 \u043a\u043b\u0438\u0435\u043d\u0442\u0430<\/p>\n<\/td>\n<\/tr>\n
                  \n
                  amount<\/p>\n<\/td>\n
                  		\n
                  int<\/p>\n<\/td>\n
                  		\n
                  \u0441\u0443\u043c\u043c\u0430 \u043f\u043b\u0430\u0442\u0435\u0436\u0430<\/p>\n<\/td>\n<\/tr>\n
                  \n
                  tmMs<\/p>\n<\/td>\n
                  		\n
                  long<\/p>\n<\/td>\n
                  		\n
                  \u0432\u0440\u0435\u043c\u044f \u0441\u0432\u0435\u0440\u0448\u0435\u043d\u0438\u044f \u043f\u043b\u0430\u0442\u0435\u0436\u0430<\/p>\n<\/td>\n<\/tr>\n<\/tbody>\n<\/table>\n<\/div>\n<\/div>\n
                  \u0411\u0443\u0434\u0435\u043c \u043f\u043e\u043b\u0443\u0447\u0430\u0442\u044c \u043e\u0431\u043d\u043e\u0432\u043b\u0435\u043d\u0438\u044f \u043f\u043e \u0432\u0441\u0435\u043c \u0442\u0430\u0431\u043b\u0438\u0446\u0430\u043c \u0447\u0435\u0440\u0435\u0437 CDC<\/a>-\u043a\u0430\u043d\u0430\u043b (\u043f\u0443\u0441\u0442\u044c \u044d\u0442\u043e \u0431\u0443\u0434\u0435\u0442 Debezium<\/a>) \u0432 \u0442\u043e\u043f\u0438\u043a Kafka<\/a> \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 Json<\/a>. \u0417\u0430\u0434\u0430\u0447\u0435\u0439 \u043d\u0430\u0448\u0435\u0439 Job<\/em>-\u044b \u0431\u0443\u0434\u0435\u0442 \u0441\u043b\u0443\u0448\u0430\u0442\u044c \u0442\u043e\u043f\u0438\u043a\u0438 \u0438 \u0432\u044b\u0434\u0430\u0432\u0430\u0442\u044c \u043e\u0431\u043d\u043e\u0432\u043b\u0435\u043d\u0438\u044f \u043f\u043e \u043a\u043b\u0438\u0435\u043d\u0442\u0443 (\u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440: \u043a\u043b\u0438\u0435\u043d\u0442 \u0441\u043c\u0435\u043d\u0438\u043b \u043a\u043e\u043c\u043f\u0430\u043d\u0438\u044e) \u0432\u043e \u0432\u043d\u0435\u0448\u043d\u044e\u044e \u0441\u0438\u0441\u0442\u0435\u043c\u0443 (\u0432 \u0440\u0430\u043c\u043a\u0430\u0445 \u0441\u0442\u0430\u0442\u044c\u0438 \u0432 \u043a\u0430\u0447\u0435\u0441\u0442\u0432\u0435 \u0432\u043d\u0435\u0448\u043d\u0435\u0439 \u0441\u0438\u0441\u0442\u0435\u043c\u044b \u0431\u0443\u0434\u0435\u043c \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c \u043a\u043e\u043d\u0441\u043e\u043b\u044c \u0432\u044b\u0432\u043e\u0434\u0430).<\/p>\n
                  \n
                  \u0420\u0438\u0441\u0443\u043d\u043e\u043a 1 — \u0421\u0445\u0435\u043c\u0430 \u0431\u0438\u0437\u043d\u0435\u0441\u0441 \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u0430<\/figcaption><\/div>\n<\/figure>\n
                  \u041d\u0430\u0447\u0438\u043d\u0430\u0435\u043c \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0438\u0440\u043e\u0432\u0430\u0442\u044c<\/h2>\n
                  \u041d\u0430\u0448 \u0442\u0435\u0445\u043d\u043e\u043b\u043e\u0433\u0438\u0447\u0435\u0441\u043a\u0438\u0439 \u0441\u0442\u0435\u043a:<\/p>\n
                    \n
                  1. \n
                    Scala<\/a> v2.12.10<\/p>\n<\/li>\n
                  2. \n
                    Flink<\/a> v1.16.0<\/p>\n<\/li>\n
                  3. \n
                    Circe<\/a> v0.14.3<\/p>\n<\/li>\n<\/ol>\n
                    \u041c\u044b \u0440\u0430\u0437\u0431\u0435\u0440\u0435\u043c 2 \u0432\u0430\u0440\u0438\u0430\u043d\u0442\u0430 \u0440\u0435\u0448\u0435\u043d\u0438\u044f:<\/p>\n
                      \n
                    1. \n
                      \u041f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 Table API<\/a><\/p>\n<\/li>\n
                    2. \n
                      \u041f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 \u043a\u043b\u0430\u0441\u0441\u0438\u0447\u0435\u0441\u043a\u043e\u0433\u043e State Processor API<\/a><\/p>\n<\/li>\n<\/ol>\n
                      \u041f\u0440\u0435\u0436\u0434\u0435 \u0447\u0435\u043c \u043f\u0438\u0441\u0430\u0442\u044c \u043e\u0441\u043d\u043e\u0432\u043d\u0443\u044e \u043b\u043e\u0433\u0438\u043a\u0443 \u043d\u0430\u043c \u043f\u043e\u0442\u0440\u0435\u0431\u0443\u0435\u0442\u0441\u044f \u0440\u0435\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u043e\u0431\u0449\u0438\u0445 \u0448\u0430\u0433\u043e\u0432, \u043a\u043e\u0442\u043e\u0440\u044b\u043c\u0438 \u043c\u044b \u0431\u0443\u0434\u0435\u043c \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c\u0441\u044f \u0432 \u043e\u0431\u0435\u0438\u0445 \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044f\u0445.<\/p>\n
                      \u041f\u0435\u0440\u0432\u043e\u0435 \u0447\u0442\u043e \u043d\u0430\u043c \u043f\u043e\u0442\u0440\u0435\u0431\u0443\u0435\u0442\u0441\u044f — \u044d\u0442\u043e \u043d\u0430\u0443\u0447\u0438\u0442\u044c\u0441\u044f \u0447\u0438\u0442\u0430\u0442\u044c \u0442\u043e\u043f\u0438\u043a\u0438 Kafka<\/em>, \u0432 \u044d\u0442\u043e\u043c \u043d\u0430\u043c \u043f\u043e\u043c\u043e\u0436\u0435\u0442 \u0443\u0436\u0435 \u0433\u043e\u0442\u043e\u0432\u044b\u0439 Flink Kafka Connector<\/a>:<\/p>\n
                      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 \/\/ \u0432 \u0441\u043b\u0443\u0447\u0430\u0435 \u043d\u0435\u0443\u0434\u0430\u0447\u0438 \u043f\u0440\u0438 \u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438, \u043e\u0448\u0438\u0431\u043a\u0430 \u043f\u0440\u043e\u0441\u0442\u043e \u0438\u0433\u043d\u043e\u0440\u0438\u0440\u0443\u0435\u0442\u0441\u044f \u0438 \u0441\u043e\u0431\u044b\u0442\u0438\u0435 \u043e\u0442\u0431\u0440\u0430\u0441\u044b\u0432\u0430\u0435\u0442\u0441\u044f (\u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u044b\u0445 \u043f\u0440\u043e\u0435\u043a\u0442\u0430\u0445 \u0442\u0430\u043a \u0434\u0435\u043b\u0430\u0442\u044c \u043d\u0435 \u0441\u0442\u043e\u0438\u0442)     obj.foreach(out.collect)   }    override def getProducedType: TypeInformation[A] = {     TypeExtractor       .getForClass(classTag[A].runtimeClass.asInstanceOf[Class[A]])       .asInstanceOf[TypeInformation[A]]   } } <\/code><\/pre>\n
                      \u0412 \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438\/\u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 \u043d\u0430\u043c \u043f\u043e\u043c\u043e\u0436\u0435\u0442 Circe<\/em>:<\/p>\n
                      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() } <\/code><\/pre>\n
                      \u041c\u044b \u043d\u0435\u043c\u043d\u043e\u0433\u043e \u0443\u043f\u0440\u043e\u0441\u0442\u0438\u043c \u0441\u0435\u0431\u0435 \u0437\u0430\u0434\u0430\u0447\u0443, \u0441\u044b\u043c\u0438\u0442\u0438\u0440\u0443\u0435\u043c \u043f\u043e\u0432\u0435\u0434\u0435\u043d\u0438\u0435 CDC<\/em>-\u043a\u0430\u043d\u0430\u043b\u0430, \u043e\u0442\u043f\u0440\u0430\u0432\u043b\u044f\u044f \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0442\u0440\u0435\u0431\u0443\u0435\u043c\u043e\u0433\u043e \u0444\u043e\u0440\u043c\u0430\u0442\u0430 \u043d\u0430\u043f\u0440\u044f\u043c\u0443\u044e \u0432 \u0442\u043e\u043f\u0438\u043a Kafka<\/em>:<\/p>\n
                      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) }  <\/code><\/pre>\n
                      \u0422\u0435\u043f\u0435\u0440\u044c \u043c\u044b \u0433\u043e\u0442\u043e\u0432\u044b \u043f\u0440\u0438\u0441\u0442\u0443\u043f\u0438\u0442\u044c \u043a \u043e\u0441\u043d\u043e\u0432\u043d\u043e\u0439 \u0447\u0430\u0441\u0442\u0438.<\/p>\n
                      \u0418\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u043d\u0438\u0435 Flink Table API<\/h2>\n
                      \u0418\u0442\u0430\u043a, \u043f\u0440\u0435\u0436\u0434\u0435 \u0432\u0441\u0435\u0433\u043e \u043d\u0430\u043c \u043d\u0443\u0436\u043d\u043e \u043f\u043e\u043b\u0443\u0447\u0438\u0442\u044c \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e\u0435 \u043e\u043a\u0440\u0443\u0436\u0435\u043d\u0438\u0435, \u043f\u043e\u0441\u043a\u043e\u043b\u044c\u043a\u0443 \u043c\u044b \u0431\u0443\u0434\u0435\u043c \u0440\u0430\u0431\u043e\u0442\u0430\u0442\u044c \u0432 \u043f\u043e\u0442\u043e\u043a\u043e\u0432\u043e\u043c \u0440\u0435\u0436\u0438\u043c\u0435, \u0437\u0430\u0440\u0443\u0447\u0438\u043c\u0441\u044f \u0442\u0430\u043a\u043e\u0432\u044b\u043c:<\/p>\n
                      implicit val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment <\/code><\/pre>\n
                      \u0414\u0430\u043b\u0435\u0435 \u043f\u043e\u043b\u0443\u0447\u0438\u043c \u043e\u043a\u0440\u0443\u0436\u0435\u043d\u0438\u0435 \u0434\u043b\u044f \u0434\u043e\u0441\u0442\u0443\u043f\u0430 \u043a Table API<\/em> :<\/p>\n
                      implicit val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env) <\/code><\/pre>\n
                      \u0424\u043e\u0440\u043c\u0438\u0440\u0443\u0435\u043c Kafka Source<\/em> \u043f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 \u043a\u043e\u0434\u0430<\/a> \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u043c\u044b \u043f\u043e\u0434\u0433\u043e\u0442\u043e\u0432\u0438\u043b\u0438 \u0432\u044b\u0448\u0435:<\/p>\n
                      val clients: DataStream[Client] = KafkaConsumerSource.configureKafkaDataSource[Client](\"clients\") <\/code><\/pre>\n
                      \u0422\u0435\u043f\u0435\u0440\u044c \u043d\u0430\u043c \u043e\u0441\u0442\u0430\u043b\u0441\u044f \u0432\u0441\u0435\u0433\u043e \u043e\u0434\u0438\u043d \u0448\u0430\u0433 \u0434\u043e \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u044f \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u0435\u043d\u0438\u044f \u0441 \u043a\u043e\u0442\u043e\u0440\u044b\u043c \u043d\u0430\u043c \u043f\u0440\u0435\u0434\u0441\u0442\u043e\u0438\u0442 \u0440\u0430\u0431\u043e\u0442\u0430\u0442\u044c — Dynamic Tables<\/a>.  \u0427\u0442\u043e\u0431\u044b \u0441\u0434\u0435\u043b\u0430\u0442\u044c \u044d\u0442\u043e\u0442 \u0448\u0430\u0433 \u043c\u0430\u043a\u0441\u0438\u043c\u0430\u043b\u044c\u043d\u043e \u043f\u0440\u043e\u0441\u0442\u044b\u043c, \u0432\u043e\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u043c\u0441\u044f \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044f\u043c\u0438 Scala<\/a> \u0438 \u0440\u0430\u0441\u0448\u0438\u0440\u0438\u043c DataStream<\/em> \u043c\u0435\u0442\u043e\u0434\u043e\u043c toStreamTable<\/em>:<\/p>\n
                      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()       )   } } <\/code><\/pre>\n
                      \u041d\u0430 \u0432\u0445\u043e\u0434 \u0444\u0443\u043d\u043a\u0446\u0438\u0438 \u043f\u043e\u0434\u0430\u0435\u0442\u0441\u044f \u0441\u043f\u0438\u0441\u043e\u043a \u0438\u043c\u0435\u043d \u043f\u043e\u043b\u0435\u0439, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u0441\u043e\u0441\u0442\u0430\u0432\u044f\u0442 \u043f\u0435\u0440\u0432\u0438\u0447\u043d\u044b\u0439 \u043a\u043b\u044e\u0447 \u043d\u0430\u0448\u0435\u0439 \u0442\u0430\u0431\u043b\u0438\u0446\u044b. \u041a\u0430\u043a \u0438 \u0432 \u0431\u043e\u043b\u044c\u0448\u0438\u043d\u0441\u0442\u0432\u0435 \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a\u043e\u0432 \u043d\u0430\u0438\u043c\u0435\u043d\u044c\u0448\u0435\u0439 \u0435\u0434\u0438\u043d\u0438\u0446\u0435\u0439 \u0434\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b \u0441 \u0442\u0430\u0431\u043b\u0438\u0446\u0430\u043c\u0438 \u044f\u0432\u043b\u044f\u0435\u0442\u0441\u044f \u0441\u0442\u0440\u043e\u043a\u0430 (Row<\/a>). \u041d\u0443\u0436\u0435\u043d \u0441\u043f\u043e\u0441\u043e\u0431 \u043f\u0440\u0435\u043e\u0431\u0440\u0430\u0437\u043e\u0432\u0430\u043d\u0438\u044f \u0432\u0441\u0435\u0445 \u043d\u0430\u0448\u0438\u0445 \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0435\u0439 \u0432 \u0442\u0438\u043f Row<\/em>, \u043f\u0440\u043e\u0434\u043e\u043b\u0436\u0438\u043c \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c\u0441\u044f \u0432\u043e\u0437\u043c\u043e\u0436\u043d\u043e\u0441\u0442\u044f\u043c\u0438 Scala<\/em>, \u043d\u0430 \u044d\u0442\u043e\u0442 \u0440\u0430\u0437 \u0442\u0435\u043c \u0447\u0442\u043e Scala<\/em> \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u0435\u0442 \u043f\u0438\u0441\u0430\u0442\u044c \u043a\u043e\u0434 \u0432 \u0444\u0443\u043d\u043a\u0446\u0438\u043e\u043d\u0430\u043b\u044c\u043d\u043e\u043c \u0441\u0442\u0438\u043b\u0435<\/a>, \u0430 \u043a\u043e\u043d\u043a\u0440\u0435\u0442\u043d\u043e \u043c\u044b \u0432\u043e\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u043c\u0441\u044f Type Class<\/a>-\u043c\u0438. \u041d\u0430\u0448 Type Class<\/em> \u0431\u0443\u0434\u0435\u0442 \u0438\u043c\u0435\u0442\u044c \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u0439 \u0432\u0438\u0434:<\/p>\n
                      sealed trait RowTransformer[A] {   def toRow(entity: A): Option[Row]   protected val fieldsName: Array[String]   implicit def typeInformation: TypeInformation[Row] } <\/code><\/pre>\n
                      \u041f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u0438\u043c \u0440\u0435\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u044e \u0434\u043b\u044f \u043e\u0434\u043d\u043e\u0439 \u0438\u0437 \u0442\u0430\u0431\u043b\u0438\u0446 (\u0434\u043b\u044f \u043e\u0441\u0442\u0430\u043b\u044c\u043d\u044b\u0445 \u0430\u043b\u0433\u043e\u0440\u0438\u0442\u043c \u0431\u0443\u0434\u0435\u0442 \u0430\u043d\u0430\u043b\u043e\u0433\u0438\u0447\u0435\u043d):<\/p>\n
                      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\") } <\/code><\/pre>\n
                      \u0422\u0435\u043f\u0435\u0440\u044c \u043c\u044b \u043c\u043e\u0436\u0435\u043c \u043f\u0440\u0435\u043e\u0431\u0440\u0430\u0437\u043e\u0432\u044b\u0432\u0430\u0442\u044c DataStream<\/code> \u0432 Table<\/code> \u043e\u0434\u043d\u0438\u043c \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u0435\u043c:<\/p>\n
                      val clients: Table = KafkaConsumerSource.configureKafkaDataSource[Client](\"clients\")   .toStreamTable(\"clientId\") <\/code><\/pre>\n
                      \u042d\u0442\u043e \u0434\u0430\u043b\u0435\u043a\u043e \u043d\u0435 \u0435\u0434\u0438\u043d\u0441\u0442\u0432\u0435\u043d\u043d\u044b\u0439 \u0441\u043f\u043e\u0441\u043e\u0431 \u0441\u043e\u0437\u0434\u0430\u043d\u0438\u044f \u0442\u0430\u0431\u043b\u0438\u0446, \u0435\u0441\u043b\u0438 \u043c\u044b \u0438\u0437\u043d\u0430\u0447\u0430\u043b\u044c\u043d\u043e \u0445\u043e\u0442\u0438\u043c \u0440\u0430\u0431\u043e\u0442\u0430\u0442\u044c \u0441 \u0434\u0430\u043d\u043d\u044b\u043c\u0438 \u043a\u0430\u043a \u0441 \u0442\u0430\u0431\u043b\u0438\u0446\u0435\u0439, \u043d\u0435\u043e\u0431\u044f\u0437\u0430\u0442\u0435\u043b\u044c\u043d\u043e \u0441\u043f\u0435\u0440\u0432\u0430 \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u044f\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 \u0432 \u0432\u0438\u0434\u0435 DataStream<\/em>:<\/p>\n
                      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\") <\/code><\/pre>\n
                      \u0418 \u043a\u043e\u043d\u0435\u0447\u043d\u043e \u0436\u0435 \u043a\u0443\u0434\u0430 \u0431\u0435\u0437 SQL<\/em>:<\/p>\n
                      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\") <\/code><\/pre>\n
                      \u041c\u044b \u043e\u043a\u043e\u043d\u0447\u0430\u0442\u0435\u043b\u044c\u043d\u043e \u043f\u043e\u0434\u0433\u043e\u0442\u043e\u0432\u0438\u043b\u0438\u0441\u044c \u043a \u043d\u0430\u043f\u0438\u0441\u0430\u043d\u0438\u044e \u043e\u0441\u043d\u043e\u0432\u043d\u043e\u0439 \u043b\u043e\u0433\u0438\u043a\u0438. \u0414\u043b\u044f \u043d\u0430\u0448\u0435\u0439 \u0437\u0430\u0434\u0430\u0447\u0438 \u0431\u0443\u0434\u0435\u043c \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c inner join<\/em>, \u0442.\u0435. \u043f\u0440\u0438 \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u0438\u0438 \u043d\u043e\u0432\u043e\u0433\u043e \u0441\u043e\u0431\u044b\u0442\u0438\u044f \u043c\u044b \u0431\u0443\u0434\u0435\u043c \u0432\u044b\u0434\u0430\u0432\u0430\u0442\u044c \u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442 \u0432\u043e \u0432\u043d\u0435\u0448\u043d\u044e\u044e \u0441\u0438\u0441\u0442\u0435\u043c\u0443<\/em> \u0442\u043e\u043b\u044c\u043a\u043e \u0432 \u0441\u043b\u0443\u0447\u0430\u0435 \u043d\u0430\u043b\u0438\u0447\u0438\u044f \u0434\u0430\u043d\u043d\u044b\u0445 \u043f\u043e \u043a\u043b\u044e\u0447\u0443 \u0432\u043e \u0432\u0441\u0435\u0445 \u0442\u0430\u0431\u043b\u0438\u0446\u0430\u0445.<\/p>\n
                      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\") <\/code><\/pre>\n
                      \n
                      \u0420\u0438\u0441\u0443\u043d\u043e\u043a 2 — Dynamic Tables<\/figcaption><\/div>\n<\/figure>\n
                      \u0418\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u043d\u0438\u0435 State Processor API<\/h2>\n
                      \u0421\u043d\u043e\u0432\u0430 \u043d\u0430\u0447\u043d\u0435\u043c \u0441 \u043e\u043a\u0440\u0443\u0436\u0435\u043d\u0438\u044f, \u0432 \u044d\u0442\u043e\u043c \u0432\u0430\u0440\u0438\u0430\u043d\u0442\u0435 \u043d\u0430\u043c \u0431\u0443\u0434\u0435\u0442 \u0434\u043e\u0441\u0442\u0430\u0442\u043e\u0447\u043d\u043e \u0442\u043e\u043b\u044c\u043a\u043e \u043e\u043a\u0440\u0443\u0436\u0435\u043d\u0438\u044f \u0434\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b \u0432 \u043f\u043e\u0442\u043e\u043a\u043e\u0432\u043e\u043c \u0440\u0435\u0436\u0438\u043c\u0435:<\/p>\n
                      implicit val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment <\/code><\/pre>\n
                      \u0412 \u043f\u0440\u043e\u0448\u043b\u043e\u043c \u0432\u0430\u0440\u0438\u0430\u043d\u0442\u0435 \u043c\u044b \u043f\u0440\u0438\u0432\u043e\u0434\u0438\u043b\u0438 \u0432\u0441\u0435 \u043d\u0430\u0448\u0438 \u0441\u0443\u0449\u043d\u043e\u0441\u0442\u0438 \u043a \u0442\u0438\u043f\u0443 Table<\/em>, \u0441\u0435\u0439\u0447\u0430\u0441 \u043d\u0430\u043c \u043d\u0443\u0436\u043d\u043e \u0441\u0434\u0435\u043b\u0430\u0442\u044c \u043d\u0435\u0447\u0442\u043e \u043f\u043e\u0445\u043e\u0436\u0435\u0435:<\/p>\n
                      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       )     }   } } <\/code><\/pre>\n
                      \u041c\u044b \u043e\u043f\u0438\u0441\u0430\u043b\u0438<\/p>\n<\/div>\n<\/div>\n<\/div>\n<\/div>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[],"tags":[],"class_list":["post-349722","post","type-post","status-publish","format-standard","hentry"],"_links":{"self":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/349722","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=349722"}],"version-history":[{"count":0,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/349722\/revisions"}],"wp:attachment":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=349722"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=349722"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=349722"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}