Мститель Зот (Zot the Avenger) находится в своём собственном мире. На экране мы видим длинноволосого 12-летнего парня, ведущего себя как дерзкий, слегка неуклюжий подросток. Образ дополняется надетой козырьком назад бейсболкой и мешковатой футболкой. Зот в размеренном ритме и атмосфере уверенности собирается рассказать нам о файтингах в своей программе Video Games and More. Игры проецируются на экран за его спиной, мы замечаем характерное размытие камеры, направленной на кинескопный телевизор. Когда он запускает Street Fighter II и начинает играть за Балрога, поступает первый звонок в эфир по громоздкому бежевому проводному телефону.

Это не ретро-стрим с Twitch и не YouTube-шоу о винтажном железе. Это передача с общественного телевидения начала 1990-х годов, выпускавшаяся в эфир аризонского канала Access Tucson почти за 20 лет до рождения Twitch.tv. [Прим. пер.: общественное телевидение (public-access television) обычно некоммерческие масс-медиа США, в которых широкая публика может создавать телевизионные программы для ограниченного вещания по специальным каналам кабельного телевидения.]

Ты покажешь код на Mortal Kombat?, спрашивает восхищённый мальчик на линии. Зот готов показать любой код для любой игровой системы. Он перестаёт играть и откидывается на спинку кресла. Ребята из аппаратной, выведите на экран компьютерную графику, говорит он, и на экране появляется чит на включение крови в Mortal Kombat для консоли Super Nintendo. В 1993 году эта игра вызвала огромную волну споров о крови и жестокости в видеоиграх. Но задолго до появления Google и даже широкополосного доступа невозмутимый Зот уже раздавал детям то, что им было нужно.

Мне нравится прозвище, которое мне дали люди: первый стример, рассказывает Зот, больше известный под именем Джей-Джей Стайлс. Когда мы созвонились с ним через Zoom, в Аризоне было два часа ночи; первый стример рассказал мне о первом задокументированном случае стриминга видеоигр. Зот поставил передо мной одно условие: прежде чем соглашаться давать интервью, он хотел меня узнать, и мы примерно час говорили о моём прошлом. Когда я сказал ему, что жил в Лос-Анджелесе, он сразу же начал рассказывать свои воспоминания о жизни на Венис-бич. Стайлс фанат этики киберпанка, технологий как великого уравнителя возможностей и важности ведения архивов. В настоящее время он сотрудничает с Internet Archive над подготовкой постоянного онлайн-хранилища для его серии передач.

Streams like teen spirit

Передача Video Games and More родилась в 1993 году, задолго до широкого распространения идеи Интернета как сети общего пользования. В первую очередь я надеялся, что начав Video Games and More, я мог сказать в эфире: привет, ребята, я поиграл в крутые игры, и мне кажется, другим они тоже могут понравиться, поэтому я расскажу вам о них, объясняет он.

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

Помню, что после первого курса по обращению с камерой я в тот же день поработал над четырьмя телешоу, которые шли одно за другим, вспоминает Стайлс. В десять лет у меня была к этому тяга, поэтому я начал работать с молодыми людьми, тинейджерами над передачами типа The Forbidden Zone и Spanked, над настоящим контентом в стиле поколения X. После года работы над чужими проектами, от религиозных программ до информационных шоу, Стайлс подхватил лихорадку Мира Уэйна. Что если объединить его любовь к ток-шоу в стиле Мира Уэйна с тем, чем он уже и так занимался: игрой в видеоигры с друзьями и болтовнёй о том, что ему нравится?


Самые первые два эпизода Video Games and More были созданы в кафе через дорогу от телестанции, где была прямая трансляция, которую могли пользоваться любые люди. Стайлс говорит, что они стали как бы приквелом к серии из 37 эпизодов. Кажется, это было примерно в конце 1993-го, потому что в то же время Buffalo Bills играли с Cowboys на Супербоул (он оказался почти прав Супербоул XXVII проводился 31 января 1993 года). Но когда мы впервые делали шоу в студии, то играли в EA Sports NHL '93 на Sega Genesis и Mortal Kombat Genesis, и я просто постоянно болтал. В качестве собеседника был выбран его друг Джейсон Кингман. Я работал над этой программой, редактировал её, и сказал ему: Слушай, мне нужен будет соведущий на шоу, потому что я не хочу заниматься этим в одиночку, вспоминает Стайлс. Это слишком большой стресс для 12-летнего ребёнка.

Благодаря своему возрасту Стайлс имел естественное понимание своей целевой аудитории: других детей. По его прикидкам, в то время его передачу в среднем смотрело примерно 50 людей. Когда я упомянул эпизод, в котором он сказал звонившему, что Game Boy отстойная консоль, Стайлс объяснил, что сказал это, потому что не мог играть в темноте без отдельного устройства с подсветкой и увеличивающим стеклом, которое работало от батареек. Это меня разочаровало вы же знаете, что у детей нет денег? Как мне постоянно находить для него батарейки?


В некоторых фрагментах передачи Стайлс присутствовал на экране в виде головы поверх игр, в которые он играл и которые записывал.

Стайлс рассказал и о том, как экономил на производстве шоу. Он придумал способ, позволяющий при помощи паяльника или дрели превращать обычную VHS-кассету в S-VHS-кассету, которая была чуть дороже и имела повышенное разрешение. Он даже подал заявку на минигрант от Access Tucson на бесплатные S-VHS-кассеты. Как только Стайлс накопил достаточно кассет для записи личных копий каждого из эпизодов, он начал выпускать повторы эфиров.

Обычно для эпизода он записывал на VHS-кассету, как играет в игру, а затем делал монтаж на оборудовании телестанции. Особенно он гордится логотипом Zot the Avenger, который сделал сам в Deluxe Paint на Amiga 2000. Я запрограммировал координаты x-y-z так, что происходил переворот в 3D, рассказывает он, и я почти ощущаю его гордость через экран. Я очень горжусь тем, чего добился в телестудии. Я не полагался ни на кого, кроме членов моей команды, и обычно они попадали в программу, потому что я любил всё контролировать и отдавал им указания прямо в эфире.

Мать Стайлса подтолкнула его к покупке компьютера Apple II на остатки денег, полученных им от рекламы для McDonald's, поэтому у него появилось то, чего не было у большинства детей. Я был одним из первых пользователей диалап-интернета через провайдера AZ Starnet, предоставлявшего в моём регионе неограниченный доступ. В те времена это было очень круто. Изучив HTML по компьютерным журналам, он создал свой первый веб-сайт и начал собирать онлайн чит-коды и прохождения. Другим детям приходилось ждать, пока журналы придут по почте или смотреть его шоу.


Пусть в Video Games and More и не было Twitch-чата, зато тролли могли звонить по телефону (с 27:14).

Сам формат шоу тоже родился благодаря стремлению Стайлса к новым технологиям. Он подписался на множество журналов, в том числе на Game Players, Electronic Gaming Monthly, Nintendo Power и Sega Visions. Мой мозг был заполнен видеоиграми, новостями и прорывными технологиями. Ещё задолго до большинства детей он точно знал, чего ждать Sega CD и 32x, знаменитого шлема виртуальной реальности для Sega. И в некоторых эпизодах наряду с рассказами об играх, которые ему понравились, и прохождениями в прямом эфире он также зачитывал новости о железе.

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

С большой силой Зоту пришла и большая ответственность у общественного телевидения были свои правила, хоть FCC и закрывала глаза на неприличия и сквернословие кабельного телевидения. Из-за взрослых игр наподобие Mortal Kombat видеоигры в 90-х были темой горячих обсуждений. Даже несмотря на то, что я был маленьким ребёнком, я оставался продюсером, говорит Стайлс. И мог нести ответственность за всё, что выводил в эфир. Он рассказал об ещё одном местном продюсере Лу Перфидио, провокаторе, называвшем себя Великий Сатана. Перфидио допускал мастурбацию и выполнение пирсинга на женских телах в прямом эфире, а также совершил самый тяжкий грех распивание спиртного на программе. Его шоу быстро выгнали с эфира.

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


В 1997 году история Video Games and More подошла к своему концу Стайлс стал студентом и начал изучать программирование, а также заинтересовался музыкой. В качестве своего оружия он выбрал бас-гитару, и сегодня он работает музыкальным продюсером. Для меня это не было чем-то важным. Мне не нужны были церемонии, говорит он. Он по-прежнему встречается со своим старым режиссёром Марком и они играют в пинбол, но за стриминговой культурой Джей-Джей особо не следит слишком занят созданием музыки, видео и графики.

Как бы мне ни хотелось, я больше слушаю местное радио. У меня есть программы-напоминалки для Twitch но у меня не всегда удаётся смотреть его вовремя.


Что они делают, когда не молятся о мире во всём мире? Они играют в видеоигры, говорит юный Мститель Зот о фотографии тибетских монахов.

Даже сегодня, или, может быть, в особенности сегодня Стайлс верит в мощь технологий, способных изменить мир. Всё дело в общении, и во многих смыслах стриминг является полной, тотальной заменой телевидению и радио. Он происходит не по запросу, тебе нужно ловить время, когда идёт трансляция. Очевидно, что когда Стайлс был Зотом, он относился к своей ответственности очень серьёзно и через свою работу вкладывал огромное количество энергии в любовь к играм. Сегодня Стайлс искренне верит, что использование технологий может способствовать улучшению общения, и что благодаря спортивной чести можно достичь мира во всём мире.

Очевидно, что передача Video Games and More была не просто шоу это был предшественник современного стриминга игр. Технологии и их доступность объединяли нас, вспоминает Стайлс о времени, когда он работал в Access Tucson. Никто из нас не был богат. Мы просто приходили туда и использовали для донесения своего посыла очень дорогое оборудование, которое в обычной ситуации оставалось бы для нас недоступным. И это было таким киберпанковым. Я полностью осознавал, что делал тогда. Чем больше силы даёшь ребёнку, тем более ответственным он будет. А стриминг это сила.

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

Этапы разработки:

Часть 1 проектирование
Часть 2 сборка
Часть 3 кинематика
Часть 4 математика траекторий и последовательности
Часть 5 электроника
Часть 6 переход на 3D печать
Часть 7 новый корпус, прикладное ПО и протоколы общения
Часть 8 улучшенная математика передвижения
Часть 9 завершение версии 1.00

Силовая часть

Прошлая плата была собрана из того что было в кейсах с компонентами: LM2596S и no-name дроссели. Нехорошо нужно переделать. На этот раз я решил сделать 6 канальный блок питания по одному каналу на конечность, в качестве DC-DC взял LM2678. Получилась довольно приличная плата:



Нагрузочные тесты показали хорошую эффективность. При нагрузке 4А эффективность преобразования составила 92% при 12В входном и 6.5В выходном напряжении. Один такой канал вытягивает 3 сервопривода без серьезной просадки напряжения (менее 0.2В).

Внутри гексапода плата смотрится просто шикарно никакого колхоза и висящих проводов.

Плата управления

Данная часть получила минимальные изменения. В функционале осталось все так же, были переразведены USARTы для коммуникации с камерой, перемещены транзисторы для управления светодиодами, изменены тип кнопок BOOT и RESET, ну и всё в таким духе.

Была добавлена возможность управления питанием сервоприводов, т.к. текущие приводы при потере импульса продолжают удерживать последнее положение. Тут всё просто вывод микроконтроллера подключен к выводу ENABLE микросхемы LM2678 и в случае ошибки или разряда батареи контроллер сможет выключить питание.

Трансляция видео

О да, теперь гексапод может транслировать видео на телефон, либо другое устройство где есть браузер. Сделано это на базе ESP32-CAM. Я не хотел создавать себе лишних проблем и пришлось прибегнуть к запретной технике Arduino. Да, я просто взял готовый пример с передачей кадров по HTTP, немного его допилил и всё готово.

При получении HTTP GET запроса ESP32 забирает фрейм с камеры, преобразует его в JPEG формат разрешением 640х480 и отсылает чанками по WI-FI на приложение\браузер.

На ESP нет крепежных отверстий, поэтому пришлось сделать для него корпус. Получилась очень компактная камера. К сожалению, фотографии результата нет, а вытаскивать её из корпуса не очень хочется.


Корпус собирается путем запаивания крышки и дна нагретым предметом. Данный узел не планируется когда либо разбирать, всё необходимое было выведено наружу. Прошивать можно без разборки корпуса.

Теперь немного об архитектуре. Гексапод это ходячая точка доступа WI-FI, ESP32 в данном случае настроен в качестве клиента. При подаче питания гексапод поднимает точку доступа WI-FI в течении 30-40 секунд, ESP32 в это время делает попытки подключится ней и в случае успеха передает по USARTу свой IP адрес в STM. В результате мы имеем беспроводную локальную сеть.

Такая архитектура сделана по нескольким причинам:

• STM32F373 не потянет обработку такого потока данных;
• Не нужно делать свой протокол передачи изображения. На борту есть HTTP, почему бы его не использовать сразу?;
• Прямая передача данных на устройство по воздуху, минуя STM и провода;
• Возможность трансляции видео на любое устройство с браузером, которое подключилось к гексаподу. К примеру, я могу управлять гексподом с телефона и спокойно смотреть его глазами с ноутбука. Мне показалось это очень удобным.

Для просмотра видео из браузера нужно ввести в адресную строку IP адрес камеры, который будет показан в программе управления гексподом:

Долгожданный результат

Планы на будущее

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

Привет! В этой статье я хочу рассказать, как из модного микросервисного приложения можно сделать рабочую, управляемую систему с помощью трех проверенных годами методик: на примере проекта внутренней performance-based рекламы Joom.

The Project

Проект JoomAds предлагает продавцам инструменты продвижения товаров в Joom. Для продавца процесс продвижения начинается с создания рекламной кампании, которая состоит из:

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

Рекламная кампания это одна из частей состояния рекламного товара (см. Рис. 1), в которое также входят метаданные товара (наименование, доступные варианты, принадлежность к категории, и т.д.) и данные ранжирования (например, оценки эффективности показов).

Рис. 1

JoomAds API использует это состояние при обработке запросов в поиске, каталоге или в разделе Лучшее для показа оптимального товара каждому пользователю с учетом региональной доступности и персонализации.

JoomAds API может изменять часть состояния при регистрации покупок успешно прорекламированных товаров, корректируя остаток бюджета рекламных кампаний (Рис. 1). Настройками кампаний управляет сервис кампаний JoomAds Campaign, метаданными продукта сервис Inventory, данные ранжирования расположены в хранилище аналитики (Рис. 2).

Чтобы получить полное состояние товара при сборке рекламной выдачи, необходимо сделать как минимум три запроса к внутренним сервисам платформы.

Рис. 2
JoomAds API выступает в роли медиатора данной микросервисной системы.

Pure Microservices equals Problems

Вы можете задать справедливый вопрос: Есть ли проблема в таком количестве внешних коммуникаций? Всего три вызова....

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

Быстродействие

Любые внешние коммуникации (например, поход за метаданными товара в Inventory) это дополнительные накладные расходы, увеличивающие время ответа медиатора. Такие расходы не проблема на ранних этапах развития проекта: последовательные походы в JoomAds Campaign, Inventory и хранилище аналитики вносили небольшой вклад во время ответа JoomAds API, т.к. количество рекламируемых товаров было небольшим, а рекламная выдача присутствовала только в разделе Лучшее.

Однако с ростом количества товаров в рекламе и подключением трафика других разделов Joom, 95-й перцентиль времени JoomAds API достиг сотен миллисекунд вместо желаемых десятков. Такая ситуация является результатом несоответствия текущих условий эксплуатации исходным требованиям, использованным при разработке отдельных компонентов.

Например, поиск товаров Inventory не был рассчитан на высокие частоты запросов, но он нам нужен именно таким.

Отказоустойчивость

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

Отказ любой зависимости JoomAds API ведет к некорректной или неповторяемой рекламной выдаче, либо к ее полному отсутствию.

Сложность поддержки

Микросервисная архитектура позволяет снизить сложность поддержки узкоспециализированных приложений, таких как Inventory, но значительно усложняет разработку приложений-медиаторов, таких как JoomAds API.

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

Осложняется отладка приложений-медиаторов в процессе разработки, т.к. для этого требуются либо репрезентативные копии микросервисов, либо точные mock-объекты, поддержка которых тоже требует времени.

Эти наблюдения привели нас к осознанию необходимости изменений. Новый JoomAds должен генерировать экономически эффективную и согласованную рекламную выдачу при отказе JoomAds Campaign, Inventory или хранилища аналитики, а также иметь предсказуемое быстродействие и отвечать на входящие запросы быстрее 100 мс в 95% случаев.

Важно отметить, что мы считаем рекламную выдачу согласованной, если все представленные в ней товары доступны для покупки и отвечают заданным настройкам таргетирования.

Влияние негативных факторов можно снизить защитными техниками: таймауты для запросов внешних сервисов и Circuit Breaker механика, кэширование компонентов состояния товара и готовой рекламной выдачи.

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

Monolith over microservices (kind of)

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

Время ответа монолита определяется быстродействием хранилища, содержащего состояние рабочих сущностей, а также задержками, обусловленными обработкой данных.

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

Materialization

Совместить лучшие качества микросервисов и монолитной архитектуры нам позволил подход, именуемый Materialized View. Материализованные представления часто встречаются в реализациях СУБД. Основной целью их внедрения является оптимизация доступа к данным на чтение при выполнении конкретных запросов.

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

Например, для запросов состояния продукта по его идентификатору (см. Рис. 3) или запросов состояния множества продуктов по идентификатору рекламной кампании.

Рис. 3

Материализованное представление данных расположено во внутреннем хранилище JoomAds API, поэтому замыкание входящих коммуникаций на него положительно сказывается на производительности и отказоустойчивости системы, т.к. доступ на чтение теперь зависит только от доступности / производительности хранилища данных JoomAds, а не от аналогичных характеристик внешних ресурсов. JoomAds API является надежным монолитным приложением!

Но как обновлять данные Materialized View?

Data Sourcing

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

• Изолировать клиентскую сторону от проблем доступа к внешним ресурсам.
• Учитывать возможность высокого времени ответа компонентов инфраструктуры JoomAds.
• Предоставлять механизм восстановления на случай утраты текущего состояния Materialized View.

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

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

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

Обработка набора событий конкретного товара в порядке их поступления дает одинаковый результат при идентичном нулевом состоянии это свойство отлично подходит для реализации плана восстановления Materialized View в случае нештатных ситуаций.

Event Sourcing

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

В результате адаптации Event Sourcing подхода в инфраструктуре JoomAds появились три новых компонента: хранилище материализованного представления (MAT View Storage), конвейер материализации (Materialization Pipeline), а так же конвейер ранжирования (Ranking Pipeline), реализующий поточное вычисление потоварных score'ов ранжирования (см. Рис. 4).

Рис. 4

Discussion, Technologies

Materialized View и Event Sourcing позволили нам решить основные проблемы ранней архитектуры проекта JoomAds.

Специализированные Materialized View значительно повысили надежность и быстродействие клиентских запросов. Обновление данных с использованием Event Sourcing подхода повысило надежность коммуникации с внешними сервисами, предоставило инструменты контроля консистентности данных и позволило избавиться от неэффективных запросов к внешним ресурсам.

Однако у всех решений есть цена. Чем больше несовместимых классов запросов реализует ваше приложение, тем больше материализованных представлений вам требуется собрать. Такой подход увеличивает потребление ресурсов по памяти, системе хранения данных и CPU. Материализованные представления JoomAds располагаются в хранилище Apache Cassandra, поэтому процесс порождения новых представлений, удаления старых или модификации существующих можно назвать безболезненным.

В нашем случае MAT View целиком хранится в одной таблице Cassandra: добавление колонок в таблицы Cassandra безболезненная операция, удаление MAT View осуществляется удалением таблицы. Таким образом, крайне важно выбрать удачное хранилище для реализации Materialized View в вашем проекте.

Event Sourcing предъявляет серьезные требования к своим пользователям. Генерация событий изменения данных во всех интересующих подсистемах в заданном формате с возможностью установления хронологического порядка следования это сложная организационно-техническая задача, которую крайне трудно реализовать в середине жизненного цикла ПО. Можно назвать удачей, что хранилище данных Inventory уже имело функцию генерации событий на обновление метаданных продуктов.

Однако это лишь часть истории. Для завершения системы Event Sourcing требуется механизм регистрации событий, их упорядоченного и отказоустойчивого хранения с очисткой устаревших данных, а также процессор с поддержкой конкурентной обработки данных, управляемой отказоустойчивостью и возможностью горизонтального масштабирования. Заниматься разработкой таких систем с чистого листа сложная и трудоемкая работа, которую мы решили не проводить.

Вместо этого мы воспользовались популярными open-source решениями, развивающимися при участии Apache Software Foundation: Apache Kafka и Apache Flink.

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

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

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

Takeaway

Обмен данными на базе событий в совокупности с поточной обработкой позволяют преодолеть несоответствие контрактов различных систем, повысить контроль над отказоустойчивостью и производительностью, но требуют серьезного анализа компромиссов. Кроме того, внедрение event sourcing подхода лучше всего подходит системам, находящимся на ранних стадиях своего развития.

P.S. Этот пост был впервые опубликован в блоге Joom на vc, вы могли его встречать там. Так делать можно.

Введение

Для анализа потоковых данных необходимы источники этих данных. Так же важна сама информация, которая предоставляется источниками. А источники с текстовой информацией, к примеру, еще и редки.
Из интересных источников можно выделить следующие: twitter, vk. Но эти источники подходят не под все задачи.
Есть источники с нужными данными, но эти источники не потоковые. Здесь можно привести следующее ссылки: public-apis.
При решении задач, связанных с потоковыми данными, можно воспользоваться старым способом.
Скачать данные и отправить в поток.
Для примера можно воспользоваться следующим источником: imdb.
Следует отметить, что imdb предоставляет данные самостоятельно. См. IMDb Datasets. Но можно принять, что данные собранные напрямую содержат более актуальную информацию.

Язык: Java 1.8.
Библиотеки: kafka 2.6.0, jsoup 1.13.1.

Сбор данных

Сбор данных представляет из себя сервис, который по входным данным загружает html-страницы, ищет нужную информацию и преобразует в набор объектов.
Итак источник данных: imdb. Информация будет собираться о фильмах и будет использован следующий запрос: https://www.imdb.com/search/title/?release_date=%s,%s&countries=%s
Где 1, 2 параметр это даты. 3 параметр страны.
Для лучшего понимания источника данных можно обратится к следующему ресурсу: imdb-extensive-dataset.

Интерфейс для сервиса:

public interface MovieDirectScrapingService {    Collection<Movie> scrap();}

Класс Movie это класс, которые содержит информацию об одном фильме (или о шоу и т.п.).

class Movie {    public final String titleId;    public final String titleUrl;    public final String title;    public final String description;    public final Double rating;    public final String genres;    public final String runtime;    public final String baseUrl;    public final String baseNameUrl;    public final String baseTitleUrl;    public final String participantIds;    public final String participantNames;    public final String directorIds;    public final String directorNames;

Анализ данных на одной странице.
Информация собирается следующим образом. Данные закачиваются с помощью jsoup. Далее ищутся нужные html-элементы и трансформируются в экземпляры для фильмов.

String scrap(String url, List<Movie> items) {    Document doc = null;    try {        doc = Jsoup.connect(url).header("Accept-Language", language).get();    } catch (IOException e) {        e.printStackTrace();    }    if (doc != null) {        collectItems(doc, items);        return nextUrl(doc);    }    return "";}

Поиск ссылки на следующею страницу.

String nextUrl(Document doc) {    Elements nextPageElements = doc.select(".next-page");    if (nextPageElements.size() > 0) {        Element hrefElement = nextPageElements.get(0);        return baseUrl + hrefElement.attributes().get("href");    }    return "";}

Тогда основной метод будет таким. Формируется начальная строка поиска. Закачиваются данные по одной странице. Если есть следующая страница, то идет переход к ней. По окончании передаются накопленные данные.

@Overridepublic Collection<Movie> scrap() {    String url = String.format(            baseUrl + "/search/title/?release_date=%s,%s&countries=%s",            startDate, endDate, countries    );    List<Movie> items = new ArrayList<>();    String nextUrl = url;    while (true) {        nextUrl = scrap(nextUrl, items);        if ("".equals(nextUrl)) {            break;        }        try {            Thread.sleep(50);        } catch (InterruptedException e) {        }    }    return items;}

Подробности по остальным методам можно найти в ссылках на ресурсы.

Отправка данных в топик

Формируется следующий сервис: MovieProducer. Здесь будет один единственный публичный метод: run.

Создается продюсер для кафки. Загружаются данные из источника. Трансформируются и отправляются в топик.

public void run() {    try (SimpleStringStringProducer producer = new SimpleStringStringProducer(            bootstrapServers, clientId, topic)) {        Collection<Data.Movie> movies = movieDirectScrapingService.scrap();        List<SimpleStringStringProducer.KeyValueStringString> kvList = new ArrayList<>();        for (Data.Movie move : movies) {            Map<String, String> map = new HashMap<>();            map.put("title_id", move.titleId);            map.put("title_url", move.titleUrl);                        String value = JSONObject.toJSONString(map);            String key = UUID.randomUUID().toString();            kvList.add(new SimpleStringStringProducer.KeyValueStringString(key, value));        }        producer.produce(kvList);    }}

Теперь все вместе

Формируются нужные параметры для поиска. Загружаются данные и отправляются в топик.
Для этого понадобится еще один класс: MovieDirectScrapingExecutor. С одним публичным методом: run.

В цикле создаются данные для поиска из текущей даты. Происходит загрузка и отправка данных в топик.

public void run() {    int countriesCounter = 0;    List<String> countriesSource = Arrays.asList("us");    while (true) {        try {            LocalDate localDate = LocalDate.now();            int year = localDate.getYear();            int month = localDate.getMonthValue();            int day = localDate.getDayOfMonth();            String monthString = month < 9 ? "0" + month : Integer.toString(month);            String dayString = day < 9 ? "0" + day : Integer.toString(day);            String startDate = year + "-" + monthString + "-" + dayString;            String endDate = startDate;            String language = "en";            String countries = countriesSource.get(countriesCounter);            execute(language, startDate, endDate, countries);            Thread.sleep(1000);            countriesCounter += 1;            if (countriesCounter >= countriesSource.size()) {                countriesCounter = 0;            }        } catch (InterruptedException e) {        }    }}

Для запуска потребуется экземпляр класса MovieDirectScrapingExecutor, который можно запустить с нужными параметрами, к примеру, из метода main.

Пример отправляемых данных для одного фильма.

{  "base_name_url": "https:\/\/www.imdb.com\/name",  "participant_ids": "nm7947173~nm2373827~nm0005288~nm0942193~",  "title_id": "tt13121702",  "rating": "0.0",  "base_url": "https:\/\/www.imdb.com",  "description": "It's Christmas time and Jackie (Carly Hughes), an up-and-coming journalist, finds that her life is at a crossroads until she finds an unexpected opportunity - to run a small-town newspaper ... See full summary ",  "runtime": "",  "title": "The Christmas Edition",  "director_ids": "nm0838289~",  "title_url": "\/title\/tt13121702\/?ref_=adv_li_tt",  "director_names": "Peter Sullivan~",  "genres": "Drama, Romance",  "base_title_url": "https:\/\/www.imdb.com\/title",  "participant_names": "Carly Hughes~Rob Mayes~Marie Osmond~Aloma Wright~"}

Подробности можно найти в ссылках на ресурсы.

Тесты

Для тестирования основной логики, которая связана с отправкой данных, можно воспользоваться юнит-тестами. В тестах предварительно создается kafka-сервер.
См. Apache Kafka и тестирование с Kafka Server.

Сам тест: MovieProducerTest.

public class MovieProducerTest {    @Test    void simple() throws InterruptedException {        String brokerHost = "127.0.0.1";        int brokerPort = 29092;        String zooKeeperHost = "127.0.0.1";        int zooKeeperPort = 22183;        String bootstrapServers = brokerHost + ":" + brokerPort;        String topic = "q-data";        String clientId = "simple";        try (KafkaServerService kafkaServerService = new KafkaServerService(                brokerHost, brokerPort, zooKeeperHost, zooKeeperPort        )        ) {            kafkaServerService.start();            kafkaServerService.createTopic(topic);            MovieDirectScrapingService movieDirectScrapingServiceImpl = () -> Collections.singleton(                    new Data.Movie()            );            MovieProducer movieProducer =                    new MovieProducer(bootstrapServers, clientId, topic, movieDirectScrapingServiceImpl);            movieProducer.run();            kafkaServerService.poll(topic, "simple", 1, 5, (records) -> {                assertTrue(records.count() > 0);                ConsumerRecord<String, String> record = records.iterator().next();                JSONParser jsonParser = new JSONParser();                JSONObject jsonObject = null;                try {                    jsonObject = (JSONObject) jsonParser.parse(record.value());                } catch (ParseException e) {                    e.printStackTrace();                }                assertNotNull(jsonObject);                    });            Thread.sleep(5000);        }    }}

Заключение

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

Ссылки и ресурсы

Исходный код.

Введение

На текущий момент не так много примеров тестов для приложений на основе Spark Structured Streaming. Поэтому в данной статье приводятся базовые примеры тестов с подробным описанием.

Все примеры используют: Apache Spark 3.0.1.

Подготовка

Необходимо установить:

• Apache Spark 3.0.x
• Python 3.7 и виртуальное окружение для него
• Conda 4.y
• scikit-learn 0.22.z
• Maven 3.v
• В примерах для Scala используется версия 2.12.10.

1. Загрузить Apache Spark
2. Распаковать: tar -xvzf ./spark-3.0.1-bin-hadoop2.7.tgz
3. Создать окружение, к примеру, с помощью conda: conda create -n sp python=3.7

Необходимо настроить переменные среды. Здесь приведен пример для локального запуска.

SPARK_HOME=/Users/$USER/Documents/spark/spark-3.0.1-bin-hadoop2.7PYTHONPATH=$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.9-src.zip;

Тесты

Пример с scikit-learn

При написании тестов необходимо разделять код таким образом, чтобы можно было изолировать логику и реальное применение конечного API. Хороший пример изоляции: DataFrame-pandas, DataFrame-spark.

Для написания тестов будет использоваться следующий пример: LinearRegression.

Итак, пусть код для тестирования использует следующий "шаблон" для Python:

class XService:    def __init__(self):        # Инициализация    def train(self, ds):        # Обучение    def predict(self, ds):        # Предсказание и вывод результатов

Для Scala шаблон выглядит соответственно.

Полный пример:

from sklearn import linear_modelclass LocalService:    def __init__(self):        self.model = linear_model.LinearRegression()    def train(self, ds):        X, y = ds        self.model.fit(X, y)    def predict(self, ds):        r = self.model.predict(ds)        print(r)

Тест.

Импорт:

import unittestimport numpy as np

Основной класс:

class RunTest(unittest.TestCase):

Запуск тестов:

if __name__ == "__main__":    unittest.main()

Подготовка данных:

X = np.array([    [1, 1],  # 6    [1, 2],  # 8    [2, 2],  # 9    [2, 3]  # 11])y = np.dot(X, np.array([1, 2])) + 3  # [ 6  8  9 11], y = 1 * x_0 + 2 * x_1 + 3

Создание модели и обучение:

service = local_service.LocalService()service.train((X, y))

Получение результатов:

service.predict(np.array([[3, 5]]))service.predict(np.array([[4, 6]]))

Ответ:

[16.][19.]

Все вместе:

import unittestimport numpy as npfrom spark_streaming_pp import local_serviceclass RunTest(unittest.TestCase):    def test_run(self):        # Prepare data.        X = np.array([            [1, 1],  # 6            [1, 2],  # 8            [2, 2],  # 9            [2, 3]  # 11        ])        y = np.dot(X, np.array([1, 2])) + 3  # [ 6  8  9 11], y = 1 * x_0 + 2 * x_1 + 3        # Create model and train.        service = local_service.LocalService()        service.train((X, y))        # Predict and results.        service.predict(np.array([[3, 5]]))        service.predict(np.array([[4, 6]]))        # [16.]        # [19.]if __name__ == "__main__":    unittest.main()

Пример с Spark и Python

Будет использован аналогичный алгоритм LinearRegression. Нужно отметить, что Structured Streaming основан на тех же DataFrame-х, которые используются и в Spark Sql. Но как обычно есть нюансы.

Инициализация:

self.service = LinearRegression(maxIter=10, regParam=0.01)self.model = None

Обучение:

self.model = self.service.fit(ds)

Получение результатов:

transformed_ds = self.model.transform(ds)q = transformed_ds.select("label", "prediction").writeStream.format("console").start()return q

Все вместе:

from pyspark.ml.regression import LinearRegressionclass StructuredStreamingService:    def __init__(self):        self.service = LinearRegression(maxIter=10, regParam=0.01)        self.model = None    def train(self, ds):        self.model = self.service.fit(ds)    def predict(self, ds):        transformed_ds = self.model.transform(ds)        q = transformed_ds.select("label", "prediction").writeStream.format("console").start()        return q

Сам тест.

Обычно в тестах можно использовать данные, которые создаются прямо в тестах.

train_ds = spark.createDataFrame([    (6.0, Vectors.dense([1.0, 1.0])),    (8.0, Vectors.dense([1.0, 2.0])),    (9.0, Vectors.dense([2.0, 2.0])),    (11.0, Vectors.dense([2.0, 3.0]))],    ["label", "features"])

Это очень удобно и код получается компактным.

Но подобный код, к сожалению, не будет работать в Structured Streaming, т.к. созданный DataFrame не будет обладать нужными свойствами, хотя и будет соответствовать контракту DataFrame.
На текущий момент для создания источников для тестов можно использовать такой же подход, что и в тестах для Spark.

def test_stream_read_options_overwrite(self):    bad_schema = StructType([StructField("test", IntegerType(), False)])    schema = StructType([StructField("data", StringType(), False)])    df = self.spark.readStream.format('csv').option('path', 'python/test_support/sql/fake') \        .schema(bad_schema)\        .load(path='python/test_support/sql/streaming', schema=schema, format='text')    self.assertTrue(df.isStreaming)    self.assertEqual(df.schema.simpleString(), "struct<data:string>")

И так.

Создается контекст для работы:

spark = SparkSession.builder.enableHiveSupport().getOrCreate()spark.sparkContext.setLogLevel("ERROR")

Подготовка данных для обучения (можно сделать обычным способом):

train_ds = spark.createDataFrame([    (6.0, Vectors.dense([1.0, 1.0])),    (8.0, Vectors.dense([1.0, 2.0])),    (9.0, Vectors.dense([2.0, 2.0])),    (11.0, Vectors.dense([2.0, 3.0]))],    ["label", "features"])

Обучение:

service = structure_streaming_service.StructuredStreamingService()service.train(train_ds)

Получение результатов. Для начала считываем данные из файла и выделяем: признаки и идентификатор для объектов. После запускаем предсказание с ожиданием в 3 секунды.

def extract_features(x):    values = x.split(",")    features_ = []    for i in values[1:]:        features_.append(float(i))    features = Vectors.dense(features_)    return featuresextract_features_udf = udf(extract_features, VectorUDT())def extract_label(x):    values = x.split(",")    label = float(values[0])    return labelextract_label_udf = udf(extract_label, FloatType())predict_ds = spark.readStream.format("text").option("path", "data/structured_streaming").load() \    .withColumn("features", extract_features_udf(col("value"))) \    .withColumn("label", extract_label_udf(col("value")))service.predict(predict_ds).awaitTermination(3)

Ответ:

15.9669918.96138

Все вместе:

import unittestimport warningsfrom pyspark.sql import SparkSessionfrom pyspark.sql.functions import col, udffrom pyspark.sql.types import FloatTypefrom pyspark.ml.linalg import Vectors, VectorUDTfrom spark_streaming_pp import structure_streaming_serviceclass RunTest(unittest.TestCase):    def test_run(self):        spark = SparkSession.builder.enableHiveSupport().getOrCreate()        spark.sparkContext.setLogLevel("ERROR")        # Prepare data.        train_ds = spark.createDataFrame([            (6.0, Vectors.dense([1.0, 1.0])),            (8.0, Vectors.dense([1.0, 2.0])),            (9.0, Vectors.dense([2.0, 2.0])),            (11.0, Vectors.dense([2.0, 3.0]))        ],            ["label", "features"]        )        # Create model and train.        service = structure_streaming_service.StructuredStreamingService()        service.train(train_ds)        # Predict and results.        def extract_features(x):            values = x.split(",")            features_ = []            for i in values[1:]:                features_.append(float(i))            features = Vectors.dense(features_)            return features        extract_features_udf = udf(extract_features, VectorUDT())        def extract_label(x):            values = x.split(",")            label = float(values[0])            return label        extract_label_udf = udf(extract_label, FloatType())        predict_ds = spark.readStream.format("text").option("path", "data/structured_streaming").load() \            .withColumn("features", extract_features_udf(col("value"))) \            .withColumn("label", extract_label_udf(col("value")))        service.predict(predict_ds).awaitTermination(3)        # +-----+------------------+        # |label|        prediction|        # +-----+------------------+        # |  1.0|15.966990887541273|        # |  2.0|18.961384020443553|        # +-----+------------------+    def setUp(self):        warnings.filterwarnings("ignore", category=ResourceWarning)        warnings.filterwarnings("ignore", category=DeprecationWarning)if __name__ == "__main__":    unittest.main()

Нужно отметить, что для Scala можно воспользоваться созданием потока в памяти.
Это может выглядеть вот так:

implicit val sqlCtx = spark.sqlContextimport spark.implicits._val source = MemoryStream[Record]source.addData(Record(1.0, Vectors.dense(3.0, 5.0)))source.addData(Record(2.0, Vectors.dense(4.0, 6.0)))val predictDs = source.toDF()service.predict(predictDs).awaitTermination(2000)

Полный пример на Scala (здесь, для разнообразия, не используется sql):

package aaa.abc.dd.spark_streaming_pr.clusterimport org.apache.spark.ml.regression.{LinearRegression, LinearRegressionModel}import org.apache.spark.sql.DataFrameimport org.apache.spark.sql.functions.udfimport org.apache.spark.sql.streaming.StreamingQueryclass StructuredStreamingService {  var service: LinearRegression = _  var model: LinearRegressionModel = _  def train(ds: DataFrame): Unit = {    service = new LinearRegression().setMaxIter(10).setRegParam(0.01)    model = service.fit(ds)  }  def predict(ds: DataFrame): StreamingQuery = {    val m = ds.sparkSession.sparkContext.broadcast(model)    def transformFun(features: org.apache.spark.ml.linalg.Vector): Double = {      m.value.predict(features)    }    val transform: org.apache.spark.ml.linalg.Vector => Double = transformFun    val toUpperUdf = udf(transform)    val predictionDs = ds.withColumn("prediction", toUpperUdf(ds("features")))    predictionDs      .writeStream      .foreachBatch((r: DataFrame, i: Long) => {        r.show()        // scalastyle:off println        println(s"$i")        // scalastyle:on println      })      .start()  }}

Тест:

package aaa.abc.dd.spark_streaming_pr.clusterimport org.apache.spark.ml.linalg.Vectorsimport org.apache.spark.sql.SparkSessionimport org.apache.spark.sql.execution.streaming.MemoryStreamimport org.scalatest.{Matchers, Outcome, fixture}class StructuredStreamingServiceSuite extends fixture.FunSuite with Matchers {  test("run") { spark =>    // Prepare data.    val trainDs = spark.createDataFrame(Seq(      (6.0, Vectors.dense(1.0, 1.0)),      (8.0, Vectors.dense(1.0, 2.0)),      (9.0, Vectors.dense(2.0, 2.0)),      (11.0, Vectors.dense(2.0, 3.0))    )).toDF("label", "features")    // Create model and train.    val service = new StructuredStreamingService()    service.train(trainDs)    // Predict and results.    implicit val sqlCtx = spark.sqlContext    import spark.implicits._    val source = MemoryStream[Record]    source.addData(Record(1.0, Vectors.dense(3.0, 5.0)))    source.addData(Record(2.0, Vectors.dense(4.0, 6.0)))    val predictDs = source.toDF()    service.predict(predictDs).awaitTermination(2000)    // +-----+---------+------------------+    // |label| features|        prediction|    // +-----+---------+------------------+    // |  1.0|[3.0,5.0]|15.966990887541273|    // |  2.0|[4.0,6.0]|18.961384020443553|    // +-----+---------+------------------+  }  override protected def withFixture(test: OneArgTest): Outcome = {    val spark = SparkSession.builder().master("local[2]").getOrCreate()    try withFixture(test.toNoArgTest(spark))    finally spark.stop()  }  override type FixtureParam = SparkSession  case class Record(label: Double, features: org.apache.spark.ml.linalg.Vector)}

Выводы

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

Такие абстракции как DataFrame позволяют это сделать легко и просто.

При использовании Python данные придется хранить в файлах.

Ссылки и ресурсы

• Исходный код Python
• Исходный код Scala

Всем привет, меня зовут Олег, я занимаюсь компьютерным зрением в команде Видеоаналитики МТС и сегодня расскажу вам, как мы защищаем от небезопасного контента стриминговую платформу WASD.tv, в частности про детектирование порнографии в постановке задачи action recognition.




Наш кейс это стриминговый сайт для геймеров, киберспортсменов и любителей живых трансляций формата Twitch.tv. Одни пользователи транслируют развлекательный контент, а другие его смотрят. Контент может быть самый разный: игры, живая музыка, интерактивы, мукбанг, ASMR, готовка, подкасты, прочее и в принципе не ограничен ничем, кроме воображения стримера.

И правил платформы, за соблюдением которых следят модераторы.

Зачем нужно модерировать небезопасный контент? На это есть две причины. Первая это действующее российское законодательство, по которому распространение порнографии незаконно. Вторая причина user experience. Платформа ориентирована на людей всех возрастов, и мы не можем себе позволить взрослый контент на главной странице.

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

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

Нагота это необязательно про NSFW: сумо, борьба, одетые люди в латексе всё это безопасный контент, на котором открытые решения зачастую отрабатывают некорректно.

Исходя из этих соображений, мы начали смотреть, как можно решить эту задачу. Из интересных открытых решений уже несколько лет существует обученная на закрытых данных модель Open NSFW от Yahoo (имплементация на TF). Ещё есть классный открытый репозиторий Александра Кима nsfw data scraper, из которого можно получить несколько сотен тысяч изображений с реддита, imgur и вроде бы каких-то других сайтов. Изображения разбиты на пять классов: порно, хентай, эротика, нейтральный и рисунки. На основе этих данных появилось много моделей, например раз, два
Опенсорсные решения страдают от нескольких проблем в целом невысокое качество некоторых моделей, некорректное срабатывание на вышеупомянутых сложных кейсах и безопасных изображениях вроде тверкающих девушек и мемов с Рикардо Милосом, а также проблематичность доработки, потому что либо модели устаревшие и обучены на закрытых данных, либо данные очень шумные и с непредсказуемым распределением.



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

Распознавание действий

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

Как вообще решают эту задачу? В восемнадцатом году вышёл отличный обзор от qure.ai, и кажется, что с тех пор радикального прогресса в области не произошло, так что рекомендую. Более интересный ресерч на тему видео перешёл в более сложную задачу понимания и пересказа видео. Там и графовые сетки, и self-supervised learning этому даже был полностью посвящен второй день на последнем Machines Can See.

Так вот, классификация действий. История прогресса в нейросетевых моделях примерно следующая: сначала проводили обучение трехмерных сверточных сетей с нуля (С3D), затем стали пробовать свёртки с какой-нибудь рекуррентной архитектурой или механизмом внимания; в какой-то момент Андрей Карпатый предложил разными способами мержить представления с разных кадров, еще позже стандартом стало делать двуглавые модели, где на один вход подается последовательность кадров в BGR/RGB, а на другой посчитанный на них плотный оптический поток. Еще были приколы с использованием дополнительных признаков и специальных слоёв вроде NetVLAD. В итоге мы смотрели на модели, лучше всего показавшие себя на бенчмарке UCF101, где видео разбиты по 101 классу действий. Такой моделью оказалась архитектура I3D от DeepMind, она зашла лучше всего и у нас, поэтому расскажу о ней подробнее.

DeepMind I3D

Как бейзлайны мы пробовали обучать C3D и CNN-LSTM обе модели долго обучаются и медленно сходятся. Затем мы взяли I3D, и жизнь стала лучше. Это две трёхмерные сверточные сети для BGR и оптического потока, но есть особенность в отличие от предыдущих моделей, эта предобучена на ImageNet и собственном датасете от Deepmind Kinetics-700, в котором 650 тысяч клипов и 700 классов. Это обеспечивает крайне быструю сходимость модели в несколько часов к хорошему качеству.

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

Мы подаем в модель 16 кадров, а не 64. Раньше у нас был квадратный вход, но, учитывая специфику платформы, мы поменяли соотношение сторон входа на 16:9. Задача бинарная классификация, где нулевой класс это не порно, а единичный порно. Обучали с помощью SGD с моментумом, он показал себя чуть лучше Адама. Аугментации минимальные горизонтальные флипы и JPEG-компрессия. Тут ничего особенного.

Завершая тему моделей после I3D еще выходили модели EVANet Neural Architecture Search для последовательности кадров, SlowFast Networks сеть с двумя каналами с разным фреймрейтом, и статья Google AI Temporal Cycle-Consistency Learning , но мы их не исследовали.

На чём обучали-то?

Как я уже писал выше, с данными туго. Никто их публиковать не хочет, это сложно с юридической и этической точек зрения начиная от лицензий и заканчивая согласием каждого причастного к контенту лица. Датасеты, их лицензии и публикация это вообще весело. Если кто-то хочет написать об этом статью, я с удовольствием почитаю. Из значимых академических датасетов есть только бразильский NPDI, и он, к сожалению, маленький по объему, его распределение данных недостаточно разнообразно, он состоит из ключевых кадров, и процедура его получения не самая простая. А мы ведь еще и датасет из видео хотим! Пришлось собирать самостоятельно.

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

Скрапинг обеспечит нам потенциально гораздо большее разнообразие в данных, и мы можем получить разметку достаточно дёшево, сказав, что все кадры всех видео с условного порнхаба небезопасные, а все кадры всех видео с ютуба безопасные. Есть минусы это всё надо где-то хранить, из этого надо как-то собирать датасет, и самое существенное в наивной разметке порнороликов есть шум. Это как прямые ошибки: интро, сцены, где все одеты, крупные планы без половых признаков, меню хентай-игр так и элементы, на которые модель может переобучиться: логотипы, черные экраны, монтажные склейки. Такого шума несколько процентов, и в случае терабайтов видео избавиться от него дорого. Об этом мы ещё поговорим.

Второй подход это ручная сборка. Ее плюсы заключаются в том, что мы можем моделировать любое желаемое распределение данных, данные более предсказуемы, и их проще размечать просто потому, что их меньше. Но есть и минусы. Очевидно, данных при таком подходе получается меньше, и помимо этого они могут страдать от biasа сборщиков, так как он моделирует распределение и может что-то упустить.
Мы выбрали второй подход. Составили список того, что потенциально могло бы оказаться на стриминговой платформе: самые разные игры, анимация, аниме, игра на музыкальных инструментах, реакции, мемы, хайлайты стримов и попытались покрыть самые разные возможные типы небезопасного контента от чего-то обычного до трэша в духе порно с птеродактилями. Отдельно упомянули компьютерные игры, по которым часто делают 3д-хентай Overwatch, например. И начали собирать. В итоге могу выделить два инсайта.

Фетишисты неплохие сборщики данных

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

Да и ютуберы тоже

Пример раз: на ютубе есть компиляции хайлайтов стримеров, иногда они покрывают отдельный год, длятся часами и содержат под тысячу монтажных склеек, т.е. сцен. Пример два: топы игр/аниме/сериалов. Скажем, вам надо внятно объяснить нейросети, что такое аниме. При этом в Японии огромное количество студий, стиль которых прогрессирует с каждым годом. Решение скачать видео с топами аниме за отдельные годы от известного ютубера. Или вам нужно покрыть разнообразие сцен из популярной игры. Идете и качаете ролик например videogamedunkey по этой игре.

Итерации данных

У нас было несколько итераций данных. Сначала это было около ста видео хронометражом около 70 часов с наивной разметкой все кадры с порносайтов порно, всё с ютуба непорно, из которых мы более-менее равномерно сэмплировали последовательности кадров для датасета.

Обученная таким образом модель работала неплохо, но из-за шума в данных первые модели выдавали ошибки на разного рода логотипах, черных экранах и одетых девушках на черном кожаном диване ( ). Особенно сбивали с толку черные экраны со скором 0.817, но оказалось, что в данных была ошибка в одной из компиляций порно автор случайно отрендерил видео на десять минут дольше нужного, в итоге в трейне было много опасных черных экранов.

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

Как размечали почти для всех роликов использовали инструмент от OpenCV CVAT.



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

Отлично, у нас есть сырые размеченные данные! Как нам из них получить хороший датасет? Ролики разной длины, для каждой категории собраны видео разного хронометража и степени разнообразия как это всё связать воедино? Сколько сэмплов мы можем взять из датасета? Его разнообразие как-то фундаментально ограничено (как максимум количеством кадров видео), как нам понять, что мы берем лишнего?

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

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

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

Давайте будем искать монтажные склейки

Можно было использовать просто пики нормы разности соседних кадров, но мы использовали открытую сеть специально для нахождения монтажных склеек TransNet. Это дало нам два результата: первый что мы узнали, сколько у нас в принципе сцен в данных, а второй что мы узнали, какие категории данных имеют более низкое разнообразие. Докачали хентая, майнкрафта и прочего.

Теперь наша атомная единица для нарезки не целое видео, а одна сцена. Это позволяет нам собрать максимально разнообразный датасет, сбалансированный по категориям и по классам, учитывающий безопасные сцены из порнороликов. Видео группируются по папкам категорий, и сцены сэмплируются из них поровну для каждого класса. Если мы добавляем в датасет новые видео, то донарезка/удаление лишних сэмплов будет происходить по минимуму, датасет не будет повторно нарезаться с нуля. Очень удобно.

Собрали датасет из 20000 сэмплов в трейне, 2000 в валидации и 2000 в тесте, обучили модель, нам понравились метрики на тесте, отправили в продакшн.

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

Это и есть данные со звездочкой. Они позволили нам заточиться на разнообразный контент платформы и снизить нагрузку на модераторов. Теперь в основном ложные срабатывания происходят на новых играх так, мы одно время чаще ловили Death Stranding и Valorant.

Текущий датасет состоит из 30000/5000/3000 сэмплов train/val/test.



Благодаря нашим детекторам время проверки всей платформы модераторами снижается в несколько раз. Помимо порнографии мы отлавливаем наготу, логотипы телеканалов и спортивные трансляции, но это истории для другого раза.

Fin.

Видеоверсию материала можно увидеть здесь

Недавно я написал небольшую статью о стриминге DOS игр в браузере. Настало время сделать небольшой технический обзор. Проект ведется исключительно мной, поэтому я его позиционирую как любительский. Среди общедоступных технологий позволяющих сделать стриминг игр можно выделить только WebRTC на нём и построен мой сервис. Как вы уже наверное догадались он состоит из браузерной и серверной части.

Браузерная часть

Основной компонент сервиса WebRTC сервер Janus. Из коробки он предоставляет простое API для подключения к серверу, и поддержки WebRTC протокола. Поэтому, браузерная часть получилось максимально простой, в виде обертки поверх Janus API.

Серверная часть

На стороне сервера используются dosbox, ffmpeg и Janus. Все они собраны вместе в docker контейнер.

Текущая версия сервиса использует:

• Последнюю версию dosbox
• Последнюю версию ffmpeg, скомпилированную с поддержкой кодеков vp9 и opus
• Последнюю версию janus с небольшими дополнениями (о них ниже)

Стриминг звука и видео

Когда docker стартует, супервизор запускает все три программы. Dosbox запускает игру и начинает непрерывно генерировать кадры и звуки. Эти данные перенаправляются в ffmpeg, который создает два RTP стрима (звук, видео). Плагин для стриминга Janus (стандартный компонент), слушает эти стримы и генерирует WebRTC данные для браузера.

{dosbox} --> {ffmpeg} --> {janus streaming plugin} --> {browser}


Поддержка клавиатуры

Для поддержки игровых действий (нажатие кнопок) используются каналы данных WebRTC. Каждый раз когда игрок нажимает кнопку, срабатывает обработчик клавиатуры в браузере. Он отправляет текстовые команды через дата канал на сервер.

• pipe kdown когда кнопка нажата
• pipe kup когда кнопка отпущена

На сервере эти данные обрабатываются моим расширением (исходный код). Создается именованный пайп в который записываются поступающие команды. С другой стороны, dosbox постоянно читает этот пайп и добавляет события клавиатуры в очередь для обработки.

{browser} --> {janus data text channel} --> {pipe} --> {dosbox}


Каналы данных работают в упорядоченном режиме, время жизни одного пакета 16 мс. Т.е. события клавиатуру будут доставлены в том же порядке в каком и были нажаты.

Инфраструктура

Сервис запущен на платформе Amazon. Для каждого клиента создается новая задача Fargate. После старта задача получает публичный IP, который отправляется в браузер. При получении IP браузер инициирует WebRTC соединение с Janus сервером. Когда dosbox заканчивает работу, задача Fargate автоматически останавливается. Технически нет никаких ограничений на количество одновременных игроков.

{browser} --> {+fargate} --> {ip} --> {browser}
...
{browser} --> {stop} --> {-fargate}


Вместо заключения

Получилось достаточно поверхностно,

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

Обзорная статья: DOS Cloud Gaming