Блог

Я новичок в Apache Flink и пытаюсь понять некоторые рекомендации по масштабированию потоковых заданий Flink наряду с Kafka. Некоторые вопросы, на которые я не могу найти подходящие ответы, включают:

1. Сколько потоковых заданий вы можете / должны запускать? Существует ли проблема масштабируемости при запуске слишком большого количества потоков? Сколько это слишком много?
2. Если мы запустим, скажем, 2000 потоков для удовлетворения бизнес-потребностей, каков наилучший способ управления этими потоками?
3. Какой предпочтительный способ считывания потоковых данных из одного потока в другой? Можем ли мы присоединяться к потокам, выполнять непрерывные запросы и т. Д.?

Заранее спасибо за любую поддержку и приношу извинения, если эти вопросы кажутся несколько простыми, но я пытаюсь лучше разобраться в этой технологии. Я прочитал большую часть документации, но, по общему признанию, возможно, не смог объединить некоторые концепции из-за отсутствия у меня опыта в этой области. спасибо за любую помощь!

• Количество потоков не ограничено, flink будет масштабироваться в зависимости от объема памяти / процессора диспетчера заданий / диспетчера задач, используемого распараллеливания и количества слотов. Я использую YARN для управления ресурсами. Если количество подключаемых потоков велико, то нам нужно быть немного осторожными, чтобы не вся / основная часть обработки выполнялась в некоторых менеджерах задач, поскольку это замедлит процесс. В самом потоке kafka могут быть задержки или внутренние задержки из-за того, что некоторые диспетчеры задач сильно загружены, и для этого необходимо провести профилактические проверки.

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

• Если, считывая один поток данных в другой, вы подразумеваете соединение двух потоков в терминологии flink, тогда мы можем соединить их по общему ключу и поддерживать состояние значения. Обратите внимание, что состояние значения поддерживается в диспетчере задач и не является общим для всех менеджеров задач. В противном случае, если вы подразумеваете объединение двух или более потоков, мы можем построить функции flatmapfunctions таким образом, чтобы данные из таких потоков поступали в стандартном формате.

  Пример объединения:

 val stream1: DataStream[UserBookingEvent] = BookingClosure.getSource(runmode).getSource(env)
      .map(new ClosureMapFunction)

val stream2: DataStream[UserBookingEvent] = BookingCancel.getSource(runmode).getSource(env)
      .map(new CancelMapFunction)

val unionStream: DataStream[UserBookingEvent] = stream1.union(stream2)

---

import org.apache.flink.api.common.functions.MapFunction
import org.json4s.jackson.JsonMethods.{parse => _, parseOpt => _}
import org.json4s.native.JsonMethods._
import org.slf4j.{Logger, LoggerFactory}

class CancelMapFunction extends MapFunction[String, Option[UserBookingEvent]] {
  override def map(in: String): Option[UserBookingEvent] = {
    val LOG: Logger = LoggerFactory.getLogger(classOf[CancelMapFunction])
    try {
      implicit lazy val formats = org.json4s.DefaultFormats

      val json = parse(in)
      ..............
     } catch {
      case e: Exception => {
        LOG.error("Could not parse Cancel Event= "   in   e.getMessage)
        None
      }
    }
  }
}