Механизмы выполнения
Visary ETL поддерживает два разных механизма выполнения: традиционный и без учета состояния.
Механизм выполнения определяет, как выполняются компоненты в рамках данной группы процессов. Эти два механизма выполнения во многом отличаются друг от друга. У каждого из них есть свои плюсы и минусы, и, как таковой, Visary ETL позволяет разработчику потока настраивать, какой механизм выполнения использовать для отдельных групп процессов. В этом разделе мы расскажем о различиях между этими двумя моделями.
Традиционный механизм выполнения
По умолчанию используется традиционный механизм выполнения. При использовании традиционного механизма каждый компонент в группе процессов планируется независимо. Очереди используются для буферизации данных между процессорами, и каждый процессор обрабатывает данные в своей очереди независимо от других процессоров. Данные, находящиеся в данной очереди, сохраняются в репозиториях Visary ETL, так что при перезапуске Visary ETL данные будут восстановлены в той очереди, в которой они находились до перезапуска.
Механизм выполнения без сохранения состояния
Когда группа процессов настроена на использование механизма выполнения без сохранения состояния, вся группа процессов функционирует гораздо более аналогично процессору. Группе процессов назначается определенное количество параллельных задач. Каждая параллельная задача отвечает за выполнение всего потока. Исходные процессоры (т.е. процессоры внутри группы процессов, которые не имеют входящих подключений) будут запущены один раз. Если для группы процессов есть какие-либо входные порты, у них также будет возможность запустить и передать один потоковый файл. В этот момент поток будет выполняться до его завершения, если это возможно*, прежде чем в него будут введены какие-либо дополнительные данные.
В некоторых сценариях поток данных не может завершиться без ввода дополнительных данных. Например, процессору MergeContent может потребоваться дополнительный ввод данных, прежде чем он сможет продолжить работу. В этом случае исходным компонентам будет разрешено запускаться снова до тех пор, пока компонент не сможет продолжить работу.
Данные, находящиеся в очереди в группе без сохранения состояния, не сохраняются. В результате, если перезапустить Visary ETL, данные не будут восстановлены в очередь, в которой они находились до перезапуска. Вместо этого данные, обрабатываемые группой без учета состояния, теряются. Это открывает некоторые интересные возможности, которые будут рассмотрены ниже, в разделе "Когда использовать механизм выполнения без сохранения состояния и стандартный механизм выполнения".
Сравнение и противопоставление
Теперь, когда у нас есть общее представление о роли различных механизмов выполнения, мы сравним и сопоставим, чем каждый из механизмов выполнения отличается по нескольким важным возможностям.
Планирование
Традиционный движок планирует работу каждого процессора индивидуально. Один процессор может быть настроен на запуск каждые пять минут, в то время как другой процессор должен работать как можно быстрее, используя восемь параллельных задач. Основная задача очереди между процессорами - обеспечить работу процессоров с разной скоростью. Если источник работает быстрее, чем получатель, он будет помещать данные в очередь до тех пор, пока не будет достигнут заданный порог обратного давления. Затем источник прекратит выдачу данных, чтобы у получателя была возможность наверстать упущенное. Если в очереди слишком много потоковых файлов, они (т.е. атрибуты и т.д.) переносятся на диск, чтобы избежать исчерпания кучи Java. Такой подход дает разработчику потока очень точный контроль над тем, сколько ресурсов следует выделять на каждый этап обработки по всему конвейеру.
В отличие от этого, обработчик без учета состояния использует один поток для выполнения всего потока от начала до конца. Исходные компоненты запускаются первыми, чтобы ввести данные в поток. Если в поток загружается потоковый файл, запускается пункт назначения очереди, в которую поступили данные, до тех пор, пока не будут обработаны все содержащиеся в нем данные. В большинстве случаев это происходит за один вызов процессора. Затем запускается следующий процессор в потоке и так далее. Любое обратное давление, настроенное для очередей, игнорируется. Файлы потоков никогда не загружаются на диск. Однако обычно это не вызывает беспокойства, поскольку в потоке обычно обрабатывается только один потоковый файл за раз.
Однако при настройке механизма выполнения для использования в группе процессов разработчик потока может указать максимальное количество параллельных задач, которые будут использоваться для механизма без учета состояния. Если значение равно N, то есть больше 1, одновременно будет выполняться N копий потока без учета состояния. Для каждой копии потока будут запущены исходные компоненты, а затем следующий процессор в очереди и т.д. Важно отметить, что если N больше 1, то в потоке будет N копий каждого процессора, используемых для обработки данных. Следовательно, если процессору необходимо сохранить объекты внутри системы, такие как MergeContent, каждая копия потока имеет отдельную копию этого внутреннего состояния. В результате, если MergeContent настроен так, что для объединения требуется минимум 1000 потоковых файлов, а поток выполняется с 8 параллельными задачами, каждый из 8 потоков может содержать до 999 потоковых файлов, что в общей сложности составляет 7992 потоковых файла, без объединения каких-либо данных. Поэтому разработчикам потоков следует проявлять осторожность при использовании механизма сохранения состояния в сочетании с процессорами, которые должны выполнить некоторое условие, прежде чем продолжить.
Потоки без сохранения состояния могут также содержать исходные процессоры, которые не должны запускаться постоянно. Например, процессор GetFile может быть первым процессором в потоке. Может возникнуть желание запускать процессор только раз в 5 минут. Это может усложнить ситуацию, поскольку процессоры не планируются индивидуально. С этой целью проверяются все исходные процессоры, чтобы определить, как часто они должны выполняться по расписанию. Какой бы компонент ни имел наименьшее время выполнения (т.е. запланирован быстрее всего), он будет определять, как часто следует планировать поток без сохранения состояния. Например, если процессор ListFile настроен на запуск раз в час, а процессор ListS3 настроен на запуск раз в минуту, оба исходных процессора будут запускаться раз в минуту.
При использовании механизма без учета состояния всегда используется планировщик, управляемый таймером. В настоящее время планирование CRON не поддерживается для потоков без сохранения состояния. Если требуется планирование CRON, часто бывает достаточно использовать традиционный движок для процессора, управляемого CRON, а затем подключить его к группе процессов без сохранения состояния.
Традиционный движок позволяет настраивать продолжительность выполнения для каждого отдельного процессора. Это часто значительно повышает производительность. При использовании потока без сохранения состояния это недоступно. При использовании потока без сохранения состояния это управление выполняется автоматически в зависимости от используемых в потоке процессоров. Если какой-либо процессор в потоке требует последовательного запуска, поток без сохранения состояния не будет использовать продолжительность выполнения, превышающую один вызов. В противном случае он будет выполняться до 100 миллисекунд. Если какой-либо вызов приводит к сбою, продолжительность выполнения сокращается.
Транзакционность
Транзакционность - это, пожалуй, самое существенное различие между традиционным механизмом и механизмом без сохранения состояния. В традиционном механизме обработки данных процессор является границей транзакции. То есть процессор выполняет некоторое действие с потоковым файлом, а затем транзакция завершается. Потоковый файл передается следующему процессору в потоке, поскольку первый процессор выполняет какую-либо очистку.
Например, рассмотрим процессор, обрабатывающий сообщения из очереди JMS. При использовании традиционного механизма обработки сообщений процессор будет обрабатывать сообщения из JMS. Затем он создаст потоковый файл на основе сообщения JMS, перенесет этот потоковый файл в отношение success и зафиксирует сеанс. На этом этапе платформа обеспечит сохранение данных в базовых хранилищах и добавит потоковый файл в исходящую очередь. Наконец, пользователю JMS будет предоставлена возможность подтвердить получение сообщения.
В этот момент с потоковым файлом может произойти множество событий. Он может быть удален. Он может быть преобразован. Он может быть отправлен в два разных места назначения. Или он может остаться в потоке из-за сбоя в работе какой-либо нижестоящей системы, например базы данных. Однако, несмотря на это, сообщение уже было подтверждено и повторно доставлено не будет. В связи с этим важно, чтобы Visary ETL сохранил эти данные, чтобы после перезапуска он мог завершить обработку данных.
Одним из существенных преимуществ этой парадигмы является то, что источник не обязательно должен быть постоянным или транзакционным. Например, клиент может отправить HTTP POST-запрос на номер Visary ETL для предоставления данных. Пока Visary ETL отвечает сообщением "200 OK" или "201 создано" или подобным сообщением, можно предположить, что сообщение будет обработано в соответствии с потоком.
С другой стороны, если источник данных на самом деле является транзакционным, например, очередь JMS, это может оказаться менее желательным. Если Visary ETL будет отключен, возможно, из-за аппаратного сбоя, данные, которые уже были подтверждены, не будут доставлены повторно. Экземпляр Visary ETL владеет данными и должен быть восстановлен для завершения обработки данных.
Однако семантика меняется, когда вместо этого используется обработчик без учета состояния. Границей транзакции больше не является процессор, а становится вся группа процессов. Когда тот же поток выполняется с использованием механизма без учета состояния, пользователь JMS по-прежнему получает сообщение JMS и создает на его основе потоковый файл. Он по-прежнему передает потоковый файл в отношение "успех" и фиксирует сеанс. Однако при использовании механизма без учета состояния вместо сохранения данных в базовом хранилище и предоставления пользователю JMS возможности подтвердить получение сообщения, он просто передает потоковый файл следующему процессору. Затем этот процессор обрабатывает потоковый файл и передает его дальше. Пользователю JMS не предоставляется возможность подтвердить сообщение до тех пор, пока потоковый файл не будет успешно обработан во всем потоке.
Теперь это дает совершенно иной подход к построению потока данных. Это также предоставляет некоторые важные новые возможности. Например, у нас есть возможность указать последующему процессору, что при обработке данных произошел сбой. Для этого выходной порт может быть сконфигурирован как "Порт сбоя".
Теперь, если какой-либо потоковый файл перенаправляется на этот порт, вся транзакция отменяется. Все потоковые файлы в потоке будут удалены. Пользователь JMS не подтвердит сообщение, а вместо этого отправит обратно "отрицательное подтверждение", или NAK. В результате JMS-брокер повторно доставит сообщение позже. Основное преимущество этого подхода заключается в том, что если бы экземпляр Visary ETL был выключен (например, из-за аппаратного сбоя или переноса времени запуска экземпляра Visary ETL для запуска в другом месте облачной среды), другой экземпляр Visary ETL (или какое-либо другое программное обеспечение целиком) мог бы использовать сообщение JMS и обработать его. Нет необходимости гарантировать, что экземпляр Visary ETL сможет перезапуститься.
Кроме того, с понятием сбоев связано понятие тайм-аутов при использовании механизма без учета состояния. Одним из параметров конфигурации для группы процессов является "Тайм-аут потока без учета состояния". Это позволяет нам гарантировать, что данные будут обработаны в течение некоторого ожидаемого периода времени, и в противном случае выполнить откат всей транзакции. По умолчанию тайм-аут установлен на 1 минуту. Любые данные, которые передаются в поток без сохранения состояния, должны завершить обработку в течение 1 минуты, иначе произойдет тайм-аут, и вся транзакция будет отменена. Традиционный движок не имеет понятия о тайм-аутах для потока данных.
Мы также должны учитывать транзакционность, которая используется всякий раз, когда поток без сохранения состояния запускается с входного порта. Мы могли бы, например, использовать данные из любого количества источников с помощью группы процессов, использующих традиционный движок. Затем мы могли бы обрабатывать данные, используя поток без сохранения состояния. В этом случае, если какие-либо данные перенаправляются на порт сбоя, вся транзакция откатывается назад, и вместо этого исходный входной потоковый файл передается на порт сбоя, а не на тот потоковый файл, который вызвал сМы также должны учитывать транзакционность, которая используется всякий раз, когда поток без сохранения состояния запускается с входного порта. Мы могл бы, например, использовать данные из любого количества источников с помощью группы процессов, использующих традиционный движок. Затем мы могли бы обрабатывать данные, используя поток без сохранения состояния. В этом случае, если какие-либо данные перенаправляются на порт сбоя, вся транзакция откатывается назад, и вместо этого исходный входной потоковый файл передается на порт сбоя, а не на тот потоковый файл, который вызвал сбой. Если происходит тайм-аут, исходный потоковый файл блокируется и переносится обратно в исходную очередь, чтобы его можно было обработать снова.
Надежность данных
Ключевым преимуществом Visary ETL является надежность данных. В то время как многие продукты ETL и предложения по перемещению данных обеспечивают обработку в памяти, Visary ETL обеспечивает надежность данных. Все данные сохраняются в базовых хранилищах. После перезапуска Visary ETL продолжает обработку данных с того места, где он был прерван. Это означает, что он способен надежно работать практически с любым протоколом, даже с протоколами, для которых получатель не отправляет подтверждения на уровне приложения, например, с прямыми TCP-соединениями.
Однако, хотя это справедливо для традиционного движка, это не относится к движку без сохранения состояния. В отличие от этого, движок без сохранения состояния по своей конструкции удаляет данные при перезапуске. Это означает, что механизм сохранения состояния не является безопасным выбором, если данные должны быть сохранены с помощью Visary ETL. Однако для протоколов, которые предоставляют подтверждения на уровне приложений, таких как JMS, Apache Kafka, Amazon Kinesis и т.д. это может быть отличным выбором. Кроме того, для таких протоколов, как HTTP, Visary ETL предлагает процессоры, способные принимать данные, выполнять некоторую обработку и затем отправлять подтверждение. Таким образом, это безопасно, даже если Visary ETL отвечает за прием входящих подключений. Ключевым моментом здесь является сообщение подтверждения на уровне приложения, отправляемое с Visary ETL.
Упорядочивание данных
Visary ETL предоставляет возможность упорядочивать данные в каждой очереди с помощью определения приоритетов файлов потоков. Это позволяет разработчику потока данных определять, какие конкретные данные являются наиболее важными, и гарантировать, что они будут обработаны и/или отправлены по назначению в первую очередь. Это оказалось особенно эффективным в местах с менее надежными средствами связи, где каналы связи могут периодически отключаться и когда необходимо использовать их для передачи наиболее важных данных в первую очередь.
Однако механизм сохранения состояния предназначен для обработки данных с более высокой степенью транзакционности и, как правило, не буферизует большие объемы данных, пока адресат недоступен. Таким образом, он игнорирует настроенные средства определения приоритетов и всегда обрабатывает данные, используя очередь "Первый зашел - первый вышел" (FIFO).
Происхождение данных
При обычных обстоятельствах данные о происхождении, создаваемые потоками с использованием механизма без учета состояния, сохраняются в том же базовом хранилище данных, что и в традиционном механизме, и поэтому данные о происхождении отображаются таким же образом. Однако, поскольку механизм сохранения состояния может откатывать транзакции в случае сбоя, мы должны рассмотреть, как это влияет на источник данных. В противном случае мы могли бы показать, что содержимое было изменено, затем изменен атрибут и данные клонированы, но при перезапуске данные все равно были бы помещены в очередь за пределами обработчика без учета состояния, и это заставило бы нас поверить, что содержимое потокового файла отличается от того, что есть на самом деле.
Таким образом, события проверки подлинности не сохраняются в хранилище проверки подлинности до завершения транзакции для потока без сохранения состояния. Если потоковый файл перенаправляется на порт сбоя или если время вызова истекает, события проверки подлинности отбрасываются. Однако из этого правила есть два исключения: события SEND и REMOTE_INVOCATION. Даже если транзакция откатывается, факт отправки данных или того, что произошел какой-либо удаленный вызов, не может быть откатан. Таким образом, хранилище данных о происхождении по-прежнему обновляется, чтобы отметить тот факт, что эти события произошли.
Внутреннее состояние процессора
Когда используется обработчик без учета состояния и число одновременных задач устанавливается равным значению, превышающему единицу, каждая параллельная задача запускается с другой копией каждого процессора. Из-за этого процессоры, которым требуется сохранение внутреннего состояния, не будут совместно использовать это состояние между копиями потока. Например, если в потоке есть процессор MergeContent, который должен накопить 1000 потоковых файлов перед объединением данных, и количество одновременных задач установлено равным 3, то каждая параллельная задача может содержать до 999 потоковых файлов, что в общей сложности составляет 2997 потоковых файлов без объединения данных. Это важно учитывать при проектировании потоков, в которых процессору требуется внутреннее состояние.
Кроме того, из-за этого любой поток, состоящий из процессора, запуск которого запланирован только на Основном узле, ограничен использованием одной параллельной задачи. Установка значения выше единицы приведет к тому, что при запуске будет записываться предупреждение, и будет использоваться значение, равное единице, независимо от настроенного значения.
Обработка сбоев потока без учета состояния
Обработка сбоев с помощью традиционного механизма проста. Например, данные перенаправляются в отношение "сбой" и обрабатываются соответствующим образом. Хотя это и просто, это не всегда приводит к желаемому результату. Механизм без учета состояния совсем по-другому подходит к обработке сбоев.
Порты сбоев
Механизм сохранения состояния позволяет настроить выходной порт как "Порт сбоя", что означает, что если какой-либо потоковый файл достигает выходного порта, вся транзакция завершается сбоем. Этот выходной порт может находиться на любом уровне иерархии группы процессов. В этом случае выполняется откат всей транзакции. Это означает, что все потоковые файлы, которые в данный момент обрабатываются в рамках этой транзакции, будут удалены. Если источником транзакции является порт ввода, потоковый файл, который был введен через этот порт ввода, будет передан на указанный порт вывода при сбое. Если источником транзакции был процессор, этому процессору будет предоставлена возможность отправить отрицательное подтверждение ("NAK") или очистить любые ресурсы, по мере необходимости.
Тайм-ауты
Если поток без сохранения состояния не завершается в течение заданного периода ожидания, транзакция также откатывается назад, во многом таким же образом, как и при перенаправлении файла потока на порт сбоя. Однако есть одно очень заметное отличие. В то время как порт сбоя указывает, куда перенаправлять входящий потоковый файл, тайм-аут не указывает на такой вывод. Вместо этого входной потоковый файл, если он существует, блокируется и добавляется обратно в очередь ввода, чтобы его можно было попробовать снова после некоторой задержки.
Соображения по проектированию потока без учета состояния
Один источник, один пункт назначения
Поскольку транзакция без сохранения состояния может быть отменена, обычно рекомендуется, чтобы поток данных состоял из одного источника и одного адресата. Рассмотрим случай, когда файл потока данных получен из одного источника. Затем он должен быть отправлен трем разным адресатам. После успешной отправки двум из трех адресатам, данные не могут быть отправлены третьему получателю, поскольку он не работает для периодического обслуживания. В результате транзакция отменяется. Затем данные извлекаются во второй раз. Они снова успешно отправляются в первые два пункта назначения и снова не могут быть отправлены в третий пункт назначения. Они снова откатываются. Это может происходить много раз, и каждый раз данные отправляются в первые два пункта назначения. Или, возможно, иногда он отправляется только в первый пункт назначения, а во втором происходит сбой.
В такой ситуации поток в лучшем случае очень неэффективен, когда первые два пункта назначения являются идемпотентными. В худшем случае, первый и второй пункты назначения не являются идемпотентными, и существует очень большой неконтролируемый объем дублирования данных.
Эта проблема еще больше усугубляется, если поток состоит из нескольких адресатов. Рассмотрим, например, поток с двумя процессорами, каждый из которых отвечает за обработку данных из разных источников. Затем эти данные объединяются в один потоковый файл, который затем передается трем адресатам. Теперь, если выполняется откат данных, обработчики исходных данных могут извлекать разные данные из каждого из источников. В результате публикуемые данные не могут быть идемпотентными, поскольку каждый раз они состоят из разных данных, и соединение данных из каждого из этих источников каждый раз разное.
В результате рекомендуется сохранять потоки без сохранения состояния для одного источника и одного назначения. Как минимум, назначения, отличные от последнего, должны быть идемпотентными.
Осторожность при объединении
Поскольку каждая параллельная задача запускает отдельную копию потока без сохранения состояния, важно иметь в виду, что минимальное количество входящих потоковых файлов, необходимых для слияния (например, с помощью MergeContent или MergeRecord), может многократно превышать количество одновременных задач. Таким образом, для постановки такого количества данных в очередь может потребоваться некоторое время. Кроме того, поток без сохранения состояния имеет тайм-аут, ограничивающий продолжительность его выполнения. Из-за этого важно соблюдать осторожность при настройке ограничений для этих типов процессоров.
Всегда рекомендуется устанавливать максимальное время ожидания ячейки меньше, чем время ожидания потока без сохранения состояния, чтобы гарантировать, что поток данных не будет постоянно отключаться. Хотя верно, что постоянный тайм-аут, приводящий к откату транзакции, не должен препятствовать надлежащей доставке данных, это может вызвать ненужную задержку и использовать слишком много вычислительных ресурсов, если транзакции постоянно откатываются.
В результате рекомендуется, чтобы при использовании процессоров, объединяющих несколько потоковых файлов, минимальные ограничения были достаточно малы, чтобы время ожидания транзакций не было постоянным, или чтобы для таких процессоров был настроен максимальный срок хранения или что-то эквивалентное, чтобы они могли успешно продолжать работу даже при больших объемах данных. недоступен. Также стоит отметить, что если желательны более крупные выходные потоковые файлы, то использование меньшего числа параллельных задач с большей вероятностью приведет к увеличению выходных данных, поскольку входящие данные будут распределены по меньшему количеству копий потока.
Когда использовать механизм выполнения без сохранения состояния и стандартный механизм выполнения
До сих пор мы обсуждали разницу между этими двумя механизмами выполнения. Важно завершить разговор, указав, когда имеет смысл использовать тот или иной механизм выполнения.
Как правило, механизм без учета состояния может быть предпочтительным, когда выполняются три условия: источник является транзакционным, имеется одно место назначения и есть желание рассматривать обработку данного потокового файла несколькими процессорами как единую транзакцию.
Во-первых, источник должен быть транзакционным. Это означает, что у Visary ETL должен быть какой-то механизм для указания того, что он получил данные, а у источника данных должен быть механизм для повторной отправки данных, если он не получит это подтверждение. Это относится к большинству протоколов массового обслуживания, таких как JMS, Kafka, Amazon SQS и Kinesis. Это также относится к HTTP-запросам или запросам типа RPC. Однако это не относится к протоколам более низкого уровня, таким как прямые TCP-соединения, которые не имеют протокола более высокого уровня сверху.
Во-вторых, должен быть один пункт назначения. Как обсуждалось в разделе "Один источник, один пункт назначения" выше, поток без сохранения состояния, как правило, должен иметь один источник и один пункт назначения.
И, в-третьих, желательно обрабатывать весь поток как единую транзакцию. Если какая-либо часть входящего файла потока не обрабатывается, завершается ошибкой вся транзакция. Если это нежелательное поведение, вместо этого следует использовать традиционный механизм.
Вообще говоря, традиционный механизм обеспечивает несколько более высокую производительность, просто потому, что он позволяет размещать в очереди большее количество потоковых файлов между процессорами и выделять больше ресурсов отдельным процессорам. того, если пункт назначения недоступен, традиционный механизм позволяет помещать данные в очередь и периодически повторять попытку до тех пор, пока пункт назначения снова не станет доступным. Все преобразования и обработка, необходимые для приведения данных в состояние, в котором их можно отправлять, не должны выполняться более одного раза, поскольку Visary ETL помещает данные в очередь и восстанавливает их на соответствующем этапе потока. Кроме того, если один потоковый файл необходимо доставить нескольким адресатам, традиционный движок упрощает это, в то время как движок без учета состояния усложняет задачу.
Хотя в каждой ситуации, вероятно, необходимо учитывать и другие соображения, часто это наиболее важные концепции, которые следует учитывать.