Предлагаем вашему вниманию материалы по мотивам выступления Александра Сербула на конференции BigData Conference. Я, как автор и докладчик, текст немного отредактировал и добавил современных мыслей и актуальных проблем, поэтому надеюсь пост принесет вам как дополнительные практические полезные знания в отрасли, так и пищу для размышлений — куда податься со своими знаниями. Итак — в бой!
В моем понимании, BigData — это нечто безумное, т.е. термин есть, а извилин внутри него — нет. С одной стороны, администраторы ломают голову, куда складывать значительный объем информации. С другой стороны, это высоконагруженная разработка. Необходимо создавать кластер, искать хардкорных разработчиков, очень опытных. Кроме того, необходимо использовать высшую математику: дифференциальные вычисления, линейную алгебру, теорию вероятности, машинное обучение, графы и обучение на них, логистическую регрессию, линейный дискриминантный анализ и так далее. Как выжить в этой ситуации, какие инструменты взять, какие еще сырые (к сожалению, большинство), как принести фирме деньги. Ведь это основная цель использования BigData, а всё остальное — популизм.
MapReduce — это парадигма свертки функций для выполнения больших и сложных запросов по данным, предложенная Google. Она позволяет параллельно обрабатывать информацию в духе data parallelizm. В идеале, нужно все алгоритмы, используемые для работы с большими и не только большими данными, переделать на MapReduce. Но никто не делает это бесплатно ;-) В старой книжке, у бабушки на полке в паутине вы довольно быстро найдете хороший алгоритм кластеризации K-means и он будет надежный работать. Но вдруг, перед релизом, или когда данных окажется больше, чем вы ожидали, вы обнаружите, что этот алгоритм не работает в «параллельном режиме», не работает через MapReduce — им можно загрузить только одно ядро процессора. Поэтому вам нужно будет экстренно и заново изобрести еще один алгоритм, который умеет работать параллельно, и придумать, как он должен работать в парадигме MapReduce. А это удается далеко не всем – это удел Computer Science (на самом деле иногда удается без PhD знаний переделать алгоритм на MapReduce алгоритм методом крепкого кофе и доски с маркерами).
Мы в своих проектах начали с использования алгоритмов на платформе Hadoop MapReduce, но затем перешли на Spark, потому что он оказался более разумно и практично устроенным. Классический Hadoop MapReduce работает очень просто: выполнил задание, положил результат в файл. Взял, выполнил, положил. Spark — берет, выполняет все задания, а затем выгружает результат. На мой взгляд и не только мой, Hadoop MapReduce, как не крути — устаревший конгломерат, который постоянно и конвульсионно пытается измениться, из-за чего разработчикам и сисадминам постоянно приходится переучиваться, а бизнесу – играть в русскую рулетку с «сырыми технологиями». Но… выбора у нас почти нет (да, мы смотрели Apache Storm — но он совсем из другой области: task parallel computng).
Альтернатив Apache Spark, если честно, пока не видно. Это и самый активный open-source проект в инфраструктуре Apache, это и объект для подражания — посмотрите хотя бы на Prajna от Microsoft.
Можно кинуть в ответ Apache Tez или отыскать что-нибудь мелкое в зоопарке Apache — но, поверьте, для снижения рисков лучше использовать mainstream-технологии, которые развиваются в ногу с рынком.
Где-то рядом, не совсем из этой области, но из интересного и если очень хочется — посмотрите также на Apache Giraph и TensorFlow. И задайте вопрос: это TaskParallel или DataParallel технологии — и все станет понятно.
Мы используем технологии параллельной обработки данных примерно так. На MySQL-серверах крутятся сотни тысяч баз данных клиентов-компаний, со штатом от единиц до тысяч сотрудников. Spark используется в основном в сервисе персональных рекомендаций, о котором рассказывалось в одной из прошлых публикаций.
Cобираем события — просмотры заказов, добавления в корзину, оплаты заказов, — обрабатываем, кладем в Amazon Kinesis, обрабатываем worker'ами, сохраняем в NoSQL (DynamoDB) и, наконец, отправляем в Spark для генерации моделей и аналитики.
На первый взгляд, схема выглядит излишне усложненной, но постепенно приходит понимание для чего всё это нужно. Это — минимум необходимых инструментов, которые нужны для выживания.
Полученные результаты мы выгружаем в Apache Mahout и на выходе получаем конкретные рекомендации для клиента. Общий объем событий, регистрируемых сервисом рекомендаций, достигает десятков миллионов в день.
Пока мы интенсивно используем и развиваем алгоритмы коллаборативной фильтрации, но видим примерно такую дорожную карту по развитию алгоритмов:
• Мультимодальность
• Content-based рекомендации
• Кластеризация
• Machine learning, deep learning
• Таргеттинг
Сейчас, как никогда раньше, ценится мультимодальность — т.е. использование нескольких, разных алгоритмов, для ответа на вопрос (в нашем случае — выдачи персональной рекомендации). Недостаточно просто запустить сервис на базе Apache Mahout, это не даст никакого выигрыша перед конкурентами. Коллаборативной фильтрацией сегодня уже никого не удивишь. Нужно учитывать и облако тегов пользователя, когда он ходил по магазину и получал какую-то информацию. Кластеризация позволяет более гибко организовать таргетирование информации. Здесь, конечно, не обойтись без machine learning. Deep learning — это, простыми словами, «качественное» машинное обучение, подразумевающее очень детальное изучение проблемы машиной и, часто, использование многослойной рекуррентной нейронной сети. При грамотном применении это помогает еще больше увеличивать конверсию, более эффективно работать с клиентами.
Сегодня на рынке существует множество программных сред, средств, инструментов и продуктов для разработки и анализа данных. Спасибо opensource (да, там полно сырых глюковатых вурдалаков, но есть и отличные решения)! С одной стороны, разнообразие — это несомненное благо. С другой — процесс разработки может стать довольно хаотичным и бестолковым. Например, сначала пробуют использовать Apache Pig, когда что-то не получается, обращаются к Apache Hive. Посмотрели, поигрались, начинают писать на Spark SQL. Когда запросы начинают падать — кидаются к Impala (а там все еще хуже). Под угрозой суицида некоторые проклинают мир BigData и возвращаются к старым добрым РСУБД. Такое иногда впечатление, что создана куча игрушек для «специалистов», часто такими же «специалистами». А бизнес не понимает всех этих исканий, ему нужно зарабатывать деньги, он требует конкретику в срок.
Сегодня, пожалуй, лишь Apache Hive можно считать классическим и надежным инструментом для работы с SQL-запросами по распределенным данным, как и HDFS является классикой среди кластерных файловых систем (да, есть конечно еще GusterFS, CEPH). Тем не менее, многие переходят на Spark SQL, потому что этот фреймворк написан (как хочется верить) с учетом интересов бизнеса. Также, все более активно используются HBase, Casandra, Mahout, Spark MLLib. Однако требовать, чтобы каждый разработчик и/или сисадмин свободно ориентировался во всех этих инструментах — глупо. Это профанация. Технологии — глубокие, с кучей настроек и каждая требует глубокого погружения на месяца. Не спешите — из-за гонки за количеством неизбежно будет страдать качество.
В первую очередь хочу порекомендовать всем, кто работает или намеревается начать работать с параллельными алгоритмами и MapReduce, прочитать книгу “Mining of Massive Datasets”, находящуюся в открытом доступе. Ее надо прочитать несколько раз, с блокнотиком и карандашом, иначе каши в голове не избежать. Сначала ничего не будет скорее всего понятно (мне стало открываться раза с 3). Но это одна из базовых и, что важно, доступных для инженеров не обладающих черным поясом по математике, книга по алгоритмам работы с большими данными. В частности, глава 2.3.6 посвящена реляционной алгебре и способам проецирования ее операций на MapReduce. Очень полезный материал, по сути, здесь представлены готовые советы для разработчиков, достаточно их только внимательно реализовать.
Читая литературу, полную математических деталей, вспоминайте анекдот и улыбайтесь :-)
• DAG (directed acyclic graph) vs Hadoop MapReduce vs Hadoop Streaming. Можно написать большой SQL-запрос, представляющий собой несколько MapReduce-операций в цепочке, который будет корректно выполняться. Streaming реализован в Spark гораздо лучше, чем в Hadoop, им гораздо удобнее пользоваться и работает часто эффективнее, за счет кэширования данных в памяти.
• Spark Programming Guide + API. Документация весьма толковая и полезная.
• Можно программировать на Java. С++ считался сложным языком, но Scala гораздо … нет, не сложнее, скорее высокоумнее на мой взгляд. Scala — классный язык, несмотря на некую академическую протухшесть и неорганическую связь с живыми мертвецами с широко выпученными глазами типа Haskel. Я очень люблю Scala, но от него можно сойти с ума, а время компиляции оставляет желать лучшего себе и своим детям. Поэтому, при желании, к Spark можно подключиться и из Java, и из Python и из R.
• Удобная абстракция: Resilient Distributed Dataset (RDD). Прекрасная, просто божественная концепция из мира функционального программирования, позволяющая распараллеливать файлы огромного объема — на сотни гигабайт, или даже терабайты.
• Удобные коллекции: filter, map, flatMap. Удобный подход в Spark — коллекции и операции над ними: filter, map, flatMap. Откуда это взялось? Это пришло из функционального программирования, которое сейчас активно проповедуется Scala (и не только в нем).
Исторически сложилось так, что мы пишем в Spark на Java 7, а не на Java 8. К сожалению, в «семёрке» нет нормальной поддержки функциональных объектов, поэтому мы вынуждены заниматься садомазохизмом и создавать объекты типа PairFunction, Tuple2, TupleN. В общем готовьтесь — когда Java7 интегрируется со Scala, то получается жутко нечитаемый код. Я, конечно, немного утрирую, но в нем все перемешано и хочется надеть очки с 13 окулярами.
Если вы не хотите лезть в дебри Scala, то лучше используйте Java 8. Код получается более читаемый и короче.
Название Scala произошло от английского scalable. Это язык для разработки больших компонентных приложений, созданный учеными-математиками. У меня лично складывается впечатление, что Scala — это тренд. Тренд оживления функционального программирования (снова привет Haskel, F#, ...). Как то «вдруг» оказалось (хотя ученые об этом догадывались гораздо раньше), что обрабатывать массивы данных в парадигме Data-Parallel – удобнее в функциональном стиле, вау! :-) Spark активно использует и Scala Actors (здравствуй Erlang). Они позволяют писать простой, читаемый, однопоточный код, который выполняется на большом количестве серверов. Вы избавляетесь от рисков многопоточного программирования, которым вынуждены заниматься в Java — это сложно, долго и дорого (зато круто). Кроме того, из-за сложности многопоточного программирования возникает немало ошибок. А благодаря акторам жизнь «вдруг» упрощается.
Для развертывания в Amazon Spark предлагает нам скрипт под названием Spark-eс2. Он скачивает половину репозитория Беркли, что-то творит на Ruby под капотом (здравствуй Япония) и устанавливает кучу какого-то софта. Получившаяся система довольно хрупка, чувствительна к изменениям конфигурации машин. Также есть нарекания на логирование и обновление компонентов системы.
Какое-то время мы существовали со скриптом Spark-ec2, но оказалось, что лучше самостоятельно написать инсталлятор Spark. Кроме того, инсталлятор сможет учитывать возможность подключения новых spot-машин.
Всё это для нас болезненно, поскольку у нас нет большого штата сисадминов — мы больше программисты Если бы у нас было 30 сисадминов, я бы сказал: «Ребята, я буду программировать на Scala, а вы тут, пожалуйста, не спите ночами и занимайтесь кластерами Spark». Куда более привлекательным вариантом оказалась связка из Spark и Elastic MapReduce. Также коллеги хвалят решения со Spark от Cloudera и HortonWorks — может они вам тоже окажутся полезными.
Амазон предлагает нам не терять время и развернуть кластер Spark у них с использованием сервиса ElasticMapReduce. Тут почти все будет работать из коробки. Spark интегрирован в Yarn-кластер, есть куча софта, есть подглючивающий мониторинг, можно добавлять машины, масштабировать HDFS, менять размер кластера, увеличивать и уменьшать количество задач за счет spot-машин. Spot-машины в Amazon стоят в 5-10 раз дешевле. Мы их используем всегда, потому что это удобно, дешево и быстро.
Spark в EMR профессионально интегрирован с S3. Это правильно, ведь именно там вы скорее всего будете хранить файлы в Amazon. Мы сравнивали хранение больших файлов в S3 и HDFS, и оказалось, что скорость доступа примерно одинаковая. Тогда зачем связываться с HDFS, мучиться с кластерной файловой системой, если есть готовый сервис. Также в Elastic MapReduce к потрохам Sparlk/Hadoop можно прокинуть через ssh-туннелинг веб-админки и привыкнуть к ним (хотя я так и не привык).
Получается немного дороже, чем обычные машины, разница около 10%. При этом вы получаете «в комплекте» много готового, правда немного глючащего софта (Hue глючит больше всех) и возможность масштабировать кластер. При этом вам даже админ не нужен — вы, как разработчики, там цари и боги.
Машины здесь бывают трех типов:
• Master-машины, которые контролируют вообще весь кластер. На них установлен Spark Master.
• Core-машины, на которых развернута файловая система — HDFS. Их может быть несколько штук. Правда, рекомендуется только увеличивать количество core-машин, а не уменьшать, иначе теряются данные.
• Для всего остального используются task-машины. Это обычные Spark-серверы, на которых работают воркеры. Количество spot-машин можно свободно менять, создавая парк хоть из сотен машин.
• Spark. В предыдущих версиях образов Amazon пока не поддерживается spark.dynamicAllocation.enabled, так что вы должны сами говорить ему, сколько нужно машин для выполнения задачи. Если кластер частично простаивает, то Spark не займет оставшиеся машины для выполнения. Вы должны жестко прописать, сколько машин ему нужно. Это неудобно. Но начиная с AMI 4 эта функция уже работает.
• Hadoop/Yarn/HDFS. В Yarn-кластерах, как и в Oracle, используется множество настроек, и, по хорошему, нужен админ, который в этом очень хорошо разбирается. Но, несмотря на боль, Hadoop-кластеры уверенно решают свои задачи. Больше всего мне не нравится в Yarn и Hadoop то, что там бардак с логированием. В логи пишется абсолютно всё, настройки уровней логирования раскиданы по разным частям кластерных потрохов и потому их количество очень быстро разрастается. И нормального решения этой проблемы нет. Неплохо бы поучиться у старых добрых обитателей unix – например у mysql, apache.
• Ganglia. Это time-series софт, который строит графики по различным метрикам: нагрузка, количество задач и т.д. Помогает получить представление о состоянии кластера. Удобная вещь, но есть недостатки – «убитые» спот машины продолжают висеть и загромождать графики.
• Hive. Это поддержка команд SQL, которая работает на файлах в HDFS и S3. Неплохой инструмент, но иногда его возможностей не хватает. Используем. Но когда нужно больше — заходим в Spark и напрямую решаем задачи по реляционной алгебре.
• Pig. Мы его не используем, поэтому дать какую-то оценку затрудняюсь.
• Hbase. Вариант NoSQL, пока не используем.
• Impala. Очень интересная вещь, про которую можно написать отдельный пост.Пока производит впечатление сырого софта. Так что используйте на свой страх и риск.
• Hue. Это админка к «бигдате», написанная на Python. Ее GUI позволяет объединить вместе и Impala, и Hbase, и Pig, и Hive. То есть можно сделать свой уголок аналитика в вебе :-) Я им попользовался неделю, он стал глючить, зависать, потом открываться перестал вообще — в общем, недоделанный софт
Что такое Map? Мы берем что-то, раскидываем на ключи, а их уже раскидываем по кластеру. Тут ничего не должно упасть — алгоритмически.
Что такое Reduce? Когда в один worker собираются сгруппированные по одному ключу данные. Если хорошо запрограммировать, то можно порционно передавать в reducer все значения внутри одного ключа и ничего не будет падать. Но на практике оказалось, что Spark может упасть в разных точках – то буфера не хватило для сериализации, то воркеру памяти не хватило. И, на мой взгляд, это основная проблема. Но всё же можно аккуратно пристроиться. У нас Spark сейчас не падает, хотя добились мы этого с помощью магии.
Обязательно нужно поставить разумное Executor Memory: --executor-memory 20G, --conf spark.kryoserializer.buffer.mb=256, --conf spark.default.parallelism=128, --conf spark.storage.memoryFraction=0.1
KryoSerializer позволяет сжимать объекты (spark.serializer org.apache.spark.serializer.KryoSerializer). Без этого они потребляют гораздо больше памяти. Также не рекомендую уменьшать значение константы spark.default.parallelism=128, иначе может часто падать по памяти. Что касается memoryFraction, то мы не используем кэширование.
Допустим, вам нужно выгрузить из Spark данные в модель. Если объем велик, то это будет выполняться очень долго.
• Благодаря --driver-memory 10G вы понимаете, что выгружаете с драйвера.
• При использовании Colleсt() весь результат собирается в память в драйвере и он может упасть. Поэтому рекомендую использовать toLocalIterator(). Увы, его производительность очень очень низка. Поэтому нам пришлось написать код для сборки партиций. Кому интересно, расскажу подробнее.
Этот код — единственное, что помогло нам справиться с проблемой логирования:
Надеюсь было и полезно и интересно. Все больше и активнее в нашей жизни рулят параллельные алгоритмы на MapReduce. Их мало, их ищут, но иного выхода похоже нет (ну может что-то получится посчитать быстрее на Apache Giraph и TensorFlow и через парадигму Task-Parallel). Платформа, ставшая классикой – Hadoop MapReduce, уступает место функционально написанной и на современном языке и математиками платформе Apache Spark. Скорее всего вы будете вынуждены начать разбираться, хотя бы на уровне терминов, в обитателях Hadoop-зоопарка: Hive, Pig, HDFS, Presto, Impala. Но постоянно учиться – наше все и чтобы опережать конкурентов нужно знать больше, писать быстрее и думать – ярче. Всем удачи и с наступающим Новым Годом!
Bigdata
В моем понимании, BigData — это нечто безумное, т.е. термин есть, а извилин внутри него — нет. С одной стороны, администраторы ломают голову, куда складывать значительный объем информации. С другой стороны, это высоконагруженная разработка. Необходимо создавать кластер, искать хардкорных разработчиков, очень опытных. Кроме того, необходимо использовать высшую математику: дифференциальные вычисления, линейную алгебру, теорию вероятности, машинное обучение, графы и обучение на них, логистическую регрессию, линейный дискриминантный анализ и так далее. Как выжить в этой ситуации, какие инструменты взять, какие еще сырые (к сожалению, большинство), как принести фирме деньги. Ведь это основная цель использования BigData, а всё остальное — популизм.
MapReduce
MapReduce — это парадигма свертки функций для выполнения больших и сложных запросов по данным, предложенная Google. Она позволяет параллельно обрабатывать информацию в духе data parallelizm. В идеале, нужно все алгоритмы, используемые для работы с большими и не только большими данными, переделать на MapReduce. Но никто не делает это бесплатно ;-) В старой книжке, у бабушки на полке в паутине вы довольно быстро найдете хороший алгоритм кластеризации K-means и он будет надежный работать. Но вдруг, перед релизом, или когда данных окажется больше, чем вы ожидали, вы обнаружите, что этот алгоритм не работает в «параллельном режиме», не работает через MapReduce — им можно загрузить только одно ядро процессора. Поэтому вам нужно будет экстренно и заново изобрести еще один алгоритм, который умеет работать параллельно, и придумать, как он должен работать в парадигме MapReduce. А это удается далеко не всем – это удел Computer Science (на самом деле иногда удается без PhD знаний переделать алгоритм на MapReduce алгоритм методом крепкого кофе и доски с маркерами).
Мы в своих проектах начали с использования алгоритмов на платформе Hadoop MapReduce, но затем перешли на Spark, потому что он оказался более разумно и практично устроенным. Классический Hadoop MapReduce работает очень просто: выполнил задание, положил результат в файл. Взял, выполнил, положил. Spark — берет, выполняет все задания, а затем выгружает результат. На мой взгляд и не только мой, Hadoop MapReduce, как не крути — устаревший конгломерат, который постоянно и конвульсионно пытается измениться, из-за чего разработчикам и сисадминам постоянно приходится переучиваться, а бизнесу – играть в русскую рулетку с «сырыми технологиями». Но… выбора у нас почти нет (да, мы смотрели Apache Storm — но он совсем из другой области: task parallel computng).
Альтернатив Apache Spark, если честно, пока не видно. Это и самый активный open-source проект в инфраструктуре Apache, это и объект для подражания — посмотрите хотя бы на Prajna от Microsoft.
Можно кинуть в ответ Apache Tez или отыскать что-нибудь мелкое в зоопарке Apache — но, поверьте, для снижения рисков лучше использовать mainstream-технологии, которые развиваются в ногу с рынком.
Где-то рядом, не совсем из этой области, но из интересного и если очень хочется — посмотрите также на Apache Giraph и TensorFlow. И задайте вопрос: это TaskParallel или DataParallel технологии — и все станет понятно.
Для каких задач мы используем Spark
Мы используем технологии параллельной обработки данных примерно так. На MySQL-серверах крутятся сотни тысяч баз данных клиентов-компаний, со штатом от единиц до тысяч сотрудников. Spark используется в основном в сервисе персональных рекомендаций, о котором рассказывалось в одной из прошлых публикаций.
Cобираем события — просмотры заказов, добавления в корзину, оплаты заказов, — обрабатываем, кладем в Amazon Kinesis, обрабатываем worker'ами, сохраняем в NoSQL (DynamoDB) и, наконец, отправляем в Spark для генерации моделей и аналитики.
На первый взгляд, схема выглядит излишне усложненной, но постепенно приходит понимание для чего всё это нужно. Это — минимум необходимых инструментов, которые нужны для выживания.
Полученные результаты мы выгружаем в Apache Mahout и на выходе получаем конкретные рекомендации для клиента. Общий объем событий, регистрируемых сервисом рекомендаций, достигает десятков миллионов в день.
Пока мы интенсивно используем и развиваем алгоритмы коллаборативной фильтрации, но видим примерно такую дорожную карту по развитию алгоритмов:
• Мультимодальность
• Content-based рекомендации
• Кластеризация
• Machine learning, deep learning
• Таргеттинг
Сейчас, как никогда раньше, ценится мультимодальность — т.е. использование нескольких, разных алгоритмов, для ответа на вопрос (в нашем случае — выдачи персональной рекомендации). Недостаточно просто запустить сервис на базе Apache Mahout, это не даст никакого выигрыша перед конкурентами. Коллаборативной фильтрацией сегодня уже никого не удивишь. Нужно учитывать и облако тегов пользователя, когда он ходил по магазину и получал какую-то информацию. Кластеризация позволяет более гибко организовать таргетирование информации. Здесь, конечно, не обойтись без machine learning. Deep learning — это, простыми словами, «качественное» машинное обучение, подразумевающее очень детальное изучение проблемы машиной и, часто, использование многослойной рекуррентной нейронной сети. При грамотном применении это помогает еще больше увеличивать конверсию, более эффективно работать с клиентами.
Обратная сторона разнообразия
Сегодня на рынке существует множество программных сред, средств, инструментов и продуктов для разработки и анализа данных. Спасибо opensource (да, там полно сырых глюковатых вурдалаков, но есть и отличные решения)! С одной стороны, разнообразие — это несомненное благо. С другой — процесс разработки может стать довольно хаотичным и бестолковым. Например, сначала пробуют использовать Apache Pig, когда что-то не получается, обращаются к Apache Hive. Посмотрели, поигрались, начинают писать на Spark SQL. Когда запросы начинают падать — кидаются к Impala (а там все еще хуже). Под угрозой суицида некоторые проклинают мир BigData и возвращаются к старым добрым РСУБД. Такое иногда впечатление, что создана куча игрушек для «специалистов», часто такими же «специалистами». А бизнес не понимает всех этих исканий, ему нужно зарабатывать деньги, он требует конкретику в срок.
Сегодня, пожалуй, лишь Apache Hive можно считать классическим и надежным инструментом для работы с SQL-запросами по распределенным данным, как и HDFS является классикой среди кластерных файловых систем (да, есть конечно еще GusterFS, CEPH). Тем не менее, многие переходят на Spark SQL, потому что этот фреймворк написан (как хочется верить) с учетом интересов бизнеса. Также, все более активно используются HBase, Casandra, Mahout, Spark MLLib. Однако требовать, чтобы каждый разработчик и/или сисадмин свободно ориентировался во всех этих инструментах — глупо. Это профанация. Технологии — глубокие, с кучей настроек и каждая требует глубокого погружения на месяца. Не спешите — из-за гонки за количеством неизбежно будет страдать качество.
Что почитать
В первую очередь хочу порекомендовать всем, кто работает или намеревается начать работать с параллельными алгоритмами и MapReduce, прочитать книгу “Mining of Massive Datasets”, находящуюся в открытом доступе. Ее надо прочитать несколько раз, с блокнотиком и карандашом, иначе каши в голове не избежать. Сначала ничего не будет скорее всего понятно (мне стало открываться раза с 3). Но это одна из базовых и, что важно, доступных для инженеров не обладающих черным поясом по математике, книга по алгоритмам работы с большими данными. В частности, глава 2.3.6 посвящена реляционной алгебре и способам проецирования ее операций на MapReduce. Очень полезный материал, по сути, здесь представлены готовые советы для разработчиков, достаточно их только внимательно реализовать.
Читая литературу, полную математических деталей, вспоминайте анекдот и улыбайтесь :-)
Двое летят на воздушном шаре, попали в туман, заблудились. Вдруг их прижимает к земле, и они видят человека. Один из них кричит вниз: "Где мы?". Человек, подумав, отвечает: "Вы на воздушном шаре...". Очередным порывом ветра шар уносится ввысь. - Он идиот? - Нет, математик. - ?????? - Во-первых, он подумал, прежде чем ответить. Во-вторых, дал абсолютно точный ответ. А в-третьих, этот ответ совершенно никому не нужен.
Вкусности Apache Spark
• DAG (directed acyclic graph) vs Hadoop MapReduce vs Hadoop Streaming. Можно написать большой SQL-запрос, представляющий собой несколько MapReduce-операций в цепочке, который будет корректно выполняться. Streaming реализован в Spark гораздо лучше, чем в Hadoop, им гораздо удобнее пользоваться и работает часто эффективнее, за счет кэширования данных в памяти.
• Spark Programming Guide + API. Документация весьма толковая и полезная.
• Можно программировать на Java. С++ считался сложным языком, но Scala гораздо … нет, не сложнее, скорее высокоумнее на мой взгляд. Scala — классный язык, несмотря на некую академическую протухшесть и неорганическую связь с живыми мертвецами с широко выпученными глазами типа Haskel. Я очень люблю Scala, но от него можно сойти с ума, а время компиляции оставляет желать лучшего себе и своим детям. Поэтому, при желании, к Spark можно подключиться и из Java, и из Python и из R.
• Удобная абстракция: Resilient Distributed Dataset (RDD). Прекрасная, просто божественная концепция из мира функционального программирования, позволяющая распараллеливать файлы огромного объема — на сотни гигабайт, или даже терабайты.
• Удобные коллекции: filter, map, flatMap. Удобный подход в Spark — коллекции и операции над ними: filter, map, flatMap. Откуда это взялось? Это пришло из функционального программирования, которое сейчас активно проповедуется Scala (и не только в нем).
Java7 и Spark
Исторически сложилось так, что мы пишем в Spark на Java 7, а не на Java 8. К сожалению, в «семёрке» нет нормальной поддержки функциональных объектов, поэтому мы вынуждены заниматься садомазохизмом и создавать объекты типа PairFunction, Tuple2, TupleN. В общем готовьтесь — когда Java7 интегрируется со Scala, то получается жутко нечитаемый код. Я, конечно, немного утрирую, но в нем все перемешано и хочется надеть очки с 13 окулярами.
JavaRDD<String> combinedRowsOrdered = combinedRows.mapToPair(new PairFunction<String,String,String>() {…
public Tuple2<String,String> call( String row ) {
…return new Tuple2<String, String>…
Если вы не хотите лезть в дебри Scala, то лучше используйте Java 8. Код получается более читаемый и короче.
Еще немного о Scala
Название Scala произошло от английского scalable. Это язык для разработки больших компонентных приложений, созданный учеными-математиками. У меня лично складывается впечатление, что Scala — это тренд. Тренд оживления функционального программирования (снова привет Haskel, F#, ...). Как то «вдруг» оказалось (хотя ученые об этом догадывались гораздо раньше), что обрабатывать массивы данных в парадигме Data-Parallel – удобнее в функциональном стиле, вау! :-) Spark активно использует и Scala Actors (здравствуй Erlang). Они позволяют писать простой, читаемый, однопоточный код, который выполняется на большом количестве серверов. Вы избавляетесь от рисков многопоточного программирования, которым вынуждены заниматься в Java — это сложно, долго и дорого (зато круто). Кроме того, из-за сложности многопоточного программирования возникает немало ошибок. А благодаря акторам жизнь «вдруг» упрощается.
Spark-кластер
Для развертывания в Amazon Spark предлагает нам скрипт под названием Spark-eс2. Он скачивает половину репозитория Беркли, что-то творит на Ruby под капотом (здравствуй Япония) и устанавливает кучу какого-то софта. Получившаяся система довольно хрупка, чувствительна к изменениям конфигурации машин. Также есть нарекания на логирование и обновление компонентов системы.
Какое-то время мы существовали со скриптом Spark-ec2, но оказалось, что лучше самостоятельно написать инсталлятор Spark. Кроме того, инсталлятор сможет учитывать возможность подключения новых spot-машин.
Всё это для нас болезненно, поскольку у нас нет большого штата сисадминов — мы больше программисты Если бы у нас было 30 сисадминов, я бы сказал: «Ребята, я буду программировать на Scala, а вы тут, пожалуйста, не спите ночами и занимайтесь кластерами Spark». Куда более привлекательным вариантом оказалась связка из Spark и Elastic MapReduce. Также коллеги хвалят решения со Spark от Cloudera и HortonWorks — может они вам тоже окажутся полезными.
Amazon EMR
Амазон предлагает нам не терять время и развернуть кластер Spark у них с использованием сервиса ElasticMapReduce. Тут почти все будет работать из коробки. Spark интегрирован в Yarn-кластер, есть куча софта, есть подглючивающий мониторинг, можно добавлять машины, масштабировать HDFS, менять размер кластера, увеличивать и уменьшать количество задач за счет spot-машин. Spot-машины в Amazon стоят в 5-10 раз дешевле. Мы их используем всегда, потому что это удобно, дешево и быстро.
Spark в EMR профессионально интегрирован с S3. Это правильно, ведь именно там вы скорее всего будете хранить файлы в Amazon. Мы сравнивали хранение больших файлов в S3 и HDFS, и оказалось, что скорость доступа примерно одинаковая. Тогда зачем связываться с HDFS, мучиться с кластерной файловой системой, если есть готовый сервис. Также в Elastic MapReduce к потрохам Sparlk/Hadoop можно прокинуть через ssh-туннелинг веб-админки и привыкнуть к ним (хотя я так и не привык).
Стоимость Amazon EMR
Получается немного дороже, чем обычные машины, разница около 10%. При этом вы получаете «в комплекте» много готового, правда немного глючащего софта (Hue глючит больше всех) и возможность масштабировать кластер. При этом вам даже админ не нужен — вы, как разработчики, там цари и боги.
Типы машин
Машины здесь бывают трех типов:
• Master-машины, которые контролируют вообще весь кластер. На них установлен Spark Master.
• Core-машины, на которых развернута файловая система — HDFS. Их может быть несколько штук. Правда, рекомендуется только увеличивать количество core-машин, а не уменьшать, иначе теряются данные.
• Для всего остального используются task-машины. Это обычные Spark-серверы, на которых работают воркеры. Количество spot-машин можно свободно менять, создавая парк хоть из сотен машин.
Софт
• Spark. В предыдущих версиях образов Amazon пока не поддерживается spark.dynamicAllocation.enabled, так что вы должны сами говорить ему, сколько нужно машин для выполнения задачи. Если кластер частично простаивает, то Spark не займет оставшиеся машины для выполнения. Вы должны жестко прописать, сколько машин ему нужно. Это неудобно. Но начиная с AMI 4 эта функция уже работает.
• Hadoop/Yarn/HDFS. В Yarn-кластерах, как и в Oracle, используется множество настроек, и, по хорошему, нужен админ, который в этом очень хорошо разбирается. Но, несмотря на боль, Hadoop-кластеры уверенно решают свои задачи. Больше всего мне не нравится в Yarn и Hadoop то, что там бардак с логированием. В логи пишется абсолютно всё, настройки уровней логирования раскиданы по разным частям кластерных потрохов и потому их количество очень быстро разрастается. И нормального решения этой проблемы нет. Неплохо бы поучиться у старых добрых обитателей unix – например у mysql, apache.
• Ganglia. Это time-series софт, который строит графики по различным метрикам: нагрузка, количество задач и т.д. Помогает получить представление о состоянии кластера. Удобная вещь, но есть недостатки – «убитые» спот машины продолжают висеть и загромождать графики.
• Hive. Это поддержка команд SQL, которая работает на файлах в HDFS и S3. Неплохой инструмент, но иногда его возможностей не хватает. Используем. Но когда нужно больше — заходим в Spark и напрямую решаем задачи по реляционной алгебре.
• Pig. Мы его не используем, поэтому дать какую-то оценку затрудняюсь.
• Hbase. Вариант NoSQL, пока не используем.
• Impala. Очень интересная вещь, про которую можно написать отдельный пост.Пока производит впечатление сырого софта. Так что используйте на свой страх и риск.
• Hue. Это админка к «бигдате», написанная на Python. Ее GUI позволяет объединить вместе и Impala, и Hbase, и Pig, и Hive. То есть можно сделать свой уголок аналитика в вебе :-) Я им попользовался неделю, он стал глючить, зависать, потом открываться перестал вообще — в общем, недоделанный софт
Основные проблемы Spark, с которыми мы встречались
Падения по памяти
Что такое Map? Мы берем что-то, раскидываем на ключи, а их уже раскидываем по кластеру. Тут ничего не должно упасть — алгоритмически.
Что такое Reduce? Когда в один worker собираются сгруппированные по одному ключу данные. Если хорошо запрограммировать, то можно порционно передавать в reducer все значения внутри одного ключа и ничего не будет падать. Но на практике оказалось, что Spark может упасть в разных точках – то буфера не хватило для сериализации, то воркеру памяти не хватило. И, на мой взгляд, это основная проблема. Но всё же можно аккуратно пристроиться. У нас Spark сейчас не падает, хотя добились мы этого с помощью магии.
Обязательно нужно поставить разумное Executor Memory: --executor-memory 20G, --conf spark.kryoserializer.buffer.mb=256, --conf spark.default.parallelism=128, --conf spark.storage.memoryFraction=0.1
KryoSerializer позволяет сжимать объекты (spark.serializer org.apache.spark.serializer.KryoSerializer). Без этого они потребляют гораздо больше памяти. Также не рекомендую уменьшать значение константы spark.default.parallelism=128, иначе может часто падать по памяти. Что касается memoryFraction, то мы не используем кэширование.
Выгрузка результатов
Допустим, вам нужно выгрузить из Spark данные в модель. Если объем велик, то это будет выполняться очень долго.
• Благодаря --driver-memory 10G вы понимаете, что выгружаете с драйвера.
• При использовании Colleсt() весь результат собирается в память в драйвере и он может упасть. Поэтому рекомендую использовать toLocalIterator(). Увы, его производительность очень очень низка. Поэтому нам пришлось написать код для сборки партиций. Кому интересно, расскажу подробнее.
Логирование
Этот код — единственное, что помогло нам справиться с проблемой логирования:
export SPARK_LOCAL_DIRS="/mnt/spark,/mnt2/spark"
export SPARK_WORKER_DIR="/mnt/spark_worker"
export SPARK_WORKER_OPTS="-Dspark.worker.cleanup.enabled=true -Dspark.worker.cleanup.interval=1800 -Dspark.worker.cleanup.appDataTtl=259200"
#Worker executor stdout/stderr logs
spark.executor.logs.rolling.maxRetainedFiles 2
spark.executor.logs.rolling.strategy time
spark.executor.logs.rolling.time.interval daily
Заключение
Надеюсь было и полезно и интересно. Все больше и активнее в нашей жизни рулят параллельные алгоритмы на MapReduce. Их мало, их ищут, но иного выхода похоже нет (ну может что-то получится посчитать быстрее на Apache Giraph и TensorFlow и через парадигму Task-Parallel). Платформа, ставшая классикой – Hadoop MapReduce, уступает место функционально написанной и на современном языке и математиками платформе Apache Spark. Скорее всего вы будете вынуждены начать разбираться, хотя бы на уровне терминов, в обитателях Hadoop-зоопарка: Hive, Pig, HDFS, Presto, Impala. Но постоянно учиться – наше все и чтобы опережать конкурентов нужно знать больше, писать быстрее и думать – ярче. Всем удачи и с наступающим Новым Годом!
Комментарии (39)
rzykov
17.12.2015 10:21+1Что можно сделать на Mahout, что нельзя написать на Scala/Spark?
ffriend
17.12.2015 14:59+1Mahout в последних вресиях как раз и построен на Spark. По крайней мере они были в процессе переезда, когда я смотрел на них последний раз.
Другое дело, что Mahout — это библиотека готовых алгоритмов, а не фреймворк. Т.е. ваш вопрос звучит примерно как «что можно сделать на Apache 2, что нельзя сделать на C».rzykov
17.12.2015 16:01Коллаборативная фильтрация на Scala/Spark пишется в 100 строчек кода без всякого УГ как Mahout (вы пробовали с его документацией разбираться?), даже без MLLib.
ffriend
17.12.2015 17:23+1Не, так ради б-га, если у вас есть время, знания и силы писать свой алгоритм, то никто не запрещает. Только не всегда это возможно и не всегда имеет смысл. Мне вот сейчас нужна реализация ограниченной машины Больцмана на Spark, но, даже имея опыт её реализации для одной машины, я боюсь браться за эту задачу — долго и муторно. И я думаю, сделать свою коллаборативную фильтрацию для большинства пользователей Spark-а будет ещё дольше и ещё муторней.
AlexSerbul
17.12.2015 17:26Да откуда время то? Конечно сначала ищешь готовое. Благо MLlib довольно активно развивается. Кстати, пробовали на вкус уже www.tensorflow.org?
Для коллаборативки мы взяли из Mahout простую библиотечку Mahout taste, чуть дописали и задача была решена.ffriend
17.12.2015 17:43+1Да откуда время то? Конечно сначала ищешь готовое. Благо MLlib довольно активно развивается. Кстати, пробовали на вкус уже www.tensorflow.org?
Нет, не пробовал. Но у меня и область интереса немножко другая, так что в ближайшее время и не планирую.
AlexSerbul
19.12.2015 22:58Попалась реализация RMB на Spark, может будет полезной ссылочка: deeplearning4j.org/restrictedboltzmannmachine.html
ffriend
20.12.2015 01:48Видел, не понравилось. Вернее даже не так, я в нём по итогу так и не разробрался, т.е. документация касающаяся Spark неполна, но то, что я увидел в коде, меня как-то не впечатлило. D4J имеет поддержку CPU, GPU и Spark за счёт другой библиотеки — ND4J, которая вроде как предоставляет универсальные многомерные массивы, способные работать на всех трёх платформах. Вы пишете один раз, а он автомагически начинает работать везде.
Проблема в том, что CPU, GPU и Spark — это ну очень разные платформы. Очень часто, в т.ч. в конкретном случае с RBM, алгоритмы должны специально адаптироваться для каждой платформы. Здесь же я не нашёл ничего такого ни в коде, ни в комментариях. Возможно, я дико ошибаюсь, но, опять же, мне быстрее написать свою реализации RBM на MLlib, чем разбираться в их коде.
kirichenko
17.12.2015 12:56+1Ну в итоге-то много денег заработал компании? Или это всё так — популизм.
«А давайте возьмём эту хрень и перепишем её на spark»…AlexSerbul
17.12.2015 13:18Вы думаете что есть более эффективная и экономичная альтернатива бесплатному Apache Spark для решения аналогичных задач?
0x0FFF
17.12.2015 14:30+3Местами статью больно читать:
дифференциальные вычисления, линейную алгебру, теорию вероятности, машинное обучение, графы и обучение на них, логистическую регрессию, линейный дискриминантный анализ и так далее
Смесь из разделов математики и конкретных алгоритмов, непонятно как взаимосвязанных. К тому же дифференциальные вычисления никаким боком к большим объемам данных не относятся. Так же как и обозначенные алгоритмы — это то, что можно делать с данными (не обязательно большими), но чтобы работать с большими данными знать это сосвем не обязательно
Но вдруг, перед релизом, или когда данных окажется больше, чем вы ожидали, вы обнаружите, что этот алгоритм не работает в «параллельном режиме», не работает через MapReduce — им можно загрузить только одно ядро процессора. Поэтому вам нужно будет экстренно и заново изобрести еще один алгоритм, который умеет работать параллельно, и придумать, как он должен работать в парадигме MapReduce.
Вы это серьезно, переделывать алгоритм для работы на кластере нужно обязательно непосредственно перед релизом? Может для начала при выборе алгоритма стоит проверить, как он работает на кластере из виртуалок? Или хотя бы в локальном режиме MapReduce?
Spark — берет, выполняет все задания, а затем выгружает результат
При этом на каждом shuffle бережно складывая данные на жесткий диск и затем вычитывая их
Можно кинуть в ответ Apache Tez или отыскать что-нибудь мелкое в зоопарке Apache — но, поверьте, для снижения рисков лучше использовать mainstream-технологии, которые развиваются в ногу с рынком.
Apache Tez — это mainstream для Apache Hive. Никто в своем уме сейчас не использует Apache Hive поверх MapReduce: либо Hive+Tez, либо Impala или аналог
Полученные результаты мы выгружаем в Apache Mahout и на выходе получаем конкретные рекомендации для клиента
Зачем вам Apache Mahout и чем не понравился Spark MLlib? К слову сказать, Apache Mahout мертв чуть более чем 3 года
Deep learning — это, простыми словами, «качественное» машинное обучение, подразумевающее очень детальное изучение проблемы машиной и, часто, использование многослойной рекуррентной нейронной сети
Deep не имеет отношения к качеству и детальности проработки, а означает «глубину» (количество слоев) обучаемой сети
Также, все более активно используются HBase, Casandra, Mahout, Spark MLLib
Как я уже написал выше, Mahout мертв и имеет скорее отрицательную динамику использования. Также странно видеть в одном ряду два Key-Value хранилища и подпроект Spark для машинного обучения
DAG (directed acyclic graph) vs Hadoop MapReduce vs Hadoop Streaming.
DAG — абстракция уровня исполнения задачи в Spark, Hadoop MapReduce — фреймворк для обработки данных на кластере, Hadoop Streaming — дефолтный job MapReduce, который передает данные в виде текста стороннему приложению и получает от него результат. Как они могут быть в одном списке?
Streaming реализован в Spark гораздо лучше, чем в Hadoop, им гораздо удобнее пользоваться и работает часто эффективнее, за счет кэширования данных в памяти.
Как уже писали выше, MapReduce Streaming и Spark Streaming — вещи абсолютно разные
Удобные коллекции: filter, map, flatMap
Постойте, filter — это коллекция, вы уверены?
Master-машины, которые контролируют вообще весь кластер. На них установлен Spark Master
Такой вещи как Spark Master нет. Spark Master — это просто jar'ник Apache Spark, стартовавший с определенными параметрами. Это штука динамическая и зависит от того, в каком режиме вы запускаете Spark
Core-машины, на которых развернута файловая система — HDFS. Их может быть несколько штук. Правда, рекомендуется только увеличивать количество core-машин, а не уменьшать, иначе теряются данные.
Вы это серьезно? Слышали ли вы о таких вещах, как HDFS Node Decommission, HDFS Balancer, replication factor?
Для всего остального используются task-машины. Это обычные Spark-серверы, на которых работают воркеры
Вот это — sparc-серверы, а то, о чем вы пишете, это просто ноды вашего кластера, предназначенные для запуска процессов Spark
В Yarn-кластерах, как и в Oracle, используется множество настроек, и, по хорошему, нужен админ, который в этом очень хорошо разбирается
Да, для работы с кластером Hadoop нужны определенные знания. Но если вы пишете статью, то как бы подразумевается, что вы этими знаниями обладаете
Что такое Reduce? Когда в один worker собираются сгруппированные по одному ключу данные
В один worker… Коллеги, вы видели когда-нибудь настройку mapreduce.job.reduces? Один — это значение по умолчанию, их может быть сколько угодно. В Apache Spark же это задается дефолтным уровнем паралелизма и количеством партиций в целевом RDD (практически все трансформации принимают как параметр количество партиций в целевом RDD). При этом значение по умолчанию — не 1, а количество партиций в исходном RDD
Допустим, вам нужно выгрузить из Spark данные в модель. Если объем велик, то это будет выполняться очень долго
А сохранить в ту же HDFS и вычитать оттуда? А записать через тот же JdbcRDD в любимый MySQL?AlexSerbul
17.12.2015 14:37Не буду отвечать. Комментарий в стиле «привет от капитана Очевидность» :-)
0x0FFF
17.12.2015 14:44Вопрос скорее в том, что некоторые вещи позволительно сказать при живом выступлении — ошибиться в ряде терминов, назвать что-то неправильно и т.д. Это допускается, так как выступление — это зачастую импровизация. Но когда вы пишете статью ожидается, что вы проверите приведенные факты и высказывания перед публикацией, ведь в отличии от живого выступления времени на это у вас достаточно
AlexSerbul
17.12.2015 17:02Да все я проверил и хорошо понимаю технологию и мы ее используем не один год, вы придираетесь просто :-)
AlexSerbul
17.12.2015 17:17-1Deep не имеет отношения к качеству и детальности проработки, а означает «глубину» (количество слоев) обучаемой сети
А вы правда думаете, что DeepLearning делается только на нейронных сетях? :-)
Зачем вам Apache Mahout и чем не понравился Spark MLlib? К слову сказать, Apache Mahout мертв чуть более чем 3 года
Да сырой еще сильно MLLib. А ML (бизнес-процессы для MLLib) там вообще детский сад еще, бардак с интерфейсами полнейший.
k-means с ходу упал, неаккуратно реализован и неадекватно использует память, переписывать пришлось полностью.
ALS-коллаборативка — глючная, падает через раз. Алгоритмов мало. Работа с матрицами — пока примитивнейшая. До python scikit-learn еще далеко.
DAG — абстракция уровня исполнения задачи в Spark, Hadoop MapReduce — фреймворк для обработки данных на кластере, Hadoop Streaming — дефолтный job MapReduce, который передает данные в виде текста стороннему приложению и получает от него результат. Как они могут быть в одном списке?
Правда? Вот это да.
Как уже писали выше, MapReduce Streaming и Spark Streaming — вещи абсолютно разные
Серьезно? ;-)
Такой вещи как Spark Master нет. Spark Master — это просто jar'ник Apache Spark, стартовавший с определенными параметрами. Это штука динамическая и зависит от того, в каком режиме вы запускаете Spark
В амазоне такая вещь, извините, есть. В данном случае это машина с драйвером Spark. Это никакой не jar, а приложение с кучей открытых портов и жрущее немало памяти. Зачем называть приложение словом — jar ;-)
Вы это серьезно? Слышали ли вы о таких вещах, как HDFS Node Decommission, HDFS Balancer, replication factor?
Читайте внимательно, я про EMR. Не читаете внимательно, а пишете комментарий.
А сохранить в ту же HDFS и вычитать оттуда? А записать через тот же JdbcRDD в любимый MySQL?
Вы не поняли проблему. Что такое HDFS я прекрасно знаю. Зачем писать в MySQL тоже не пойму.ffriend
17.12.2015 17:28А вы правда думаете, что DeepLearning делается только на нейронных сетях? :-)
Думаю, да. То есть, некоторые виды сети можно, конечно, назвать марковским случайным полем, и это тоже будет правильно, но нейронными сетями они от этого быть не перестанут.AlexSerbul
17.12.2015 17:53А как же варианты байесовских сетей доверия и Q-leaning? Все на нейронках сидят?
ffriend
17.12.2015 22:15+1А как же варианты байесовских сетей доверия
Если байесовские сети доверия — это deep belief networks, то да, они одновременно являются и нейронными сетями. Если же вы имеете ввиду bayesian networks вообще, то тфу тфу тфу, нейронными сетями их ещё обозвать не успели, но и приставка deep к ним не применяется.
Q-leaning
Про Q-learning я знаю абсолютно ничего, но единственная работа, связанная с чем-то глубоким, на которую ссылается Педивикия, явно говорит про deep neural network. Если есть что-то ещё, кидайте ссылку, будем разбираться.AlexSerbul
17.12.2015 22:32Хорошо, спасибо. Я где-то читал, что deep learning это хотя в целом нейронки, но не всегда — зачем ограничивать себя одной технологией. Как вспомню где и что, обещаю скинуть.
ffriend
18.12.2015 00:00+1Deep learning — это именно нейронки по определению, т.к. «deep» относится именно к сетевым структурам, а всё, что сеть, можно назвать нейронкой (даже если вся суть взята из других видов сетей).
Другой вопрос, что есть representation learning — автокодировщики и RBM, основные составные блоки глубоких сетей. Вот их можно использовать и вне контекста нейронок, например, для рекоммендательных движков или уменьшения размерности.AlexSerbul
18.12.2015 00:09Вообще тема нейронок старая же. В ней регулярно бывают всплески, угасания, может и в этот раз шумиха с deeplearning скоро спадет — ну останется распознавание речи и образов вместо HMM
ffriend
Hive — это СУБД поверх HDFS, но никак не файловая система.
Hadoop Streaming — это утилита, которая «стримит» поток данных на воркерах через STDIN и STDOUT, тем самым позволяя писать MapReduce приложения на любом языке (не особо удобно и совсем не эффективно, но раньше иначе было нельзя). Т.е. это всё тот же классический MR, только mapper и reducer — не джавовские потоки, а внешние процессы. PySpark и SparkR, кстати, делают, по сути, то же самое. Как и Storm, впрочем.
А вот в Spark Streaming слово «streaming» относится уже именно к потенциально бесконечному потоку данных, когда непрерывно поступают новые данные, например, из Kafka или Twitter.
Боюсь, вы не поняли всей идеи map-reduce. Что отличает большие данные от маленьких? Правильно, размер и время передачи по сети. Map-reduce учит нас, что если данных действительно много, то гораздо проще не копировать их к приложению, а лучше скопировать само приложение к данным и обработать всё что можно локально. HDFS позволяет сделать именно это: вы не просто распределяете данные на много машин, вы ставите рядом с этими данными воркеров и назначаете этим воркерам задачи над своим локальным куском. Используя S3 вы возвращаетесь к началу и снова тратите время на перетягивание данных от одних серверов к другим.
Если хотите построить эффективную систему на Hadoop, то складывайте все пакетные данные в HDFS, а данные со случайным доступом — в какое-нибудь NoSQL хранилище (HBase, Cassandra, Redis — тут уже в зависимости от вариантов использования). Хранение в S3 имеет смысл только если вы каждый раз пересоздаёте кластер в EMR и уничтожаете его после использования, чтобы не платить Amazon-у лишних денег. Но тут уж сами считайте, что вам выгодней.
По той же причине, что и выше, core и task машины должны быть одними и теми же. Ноды данных и воркеры должны быть рядом.
После завершения приложения эта команда вытянет логи всех контейнеров YARN приложения. Нужно, правда, понимать, что именно у вас крутится в контейнерах, а что нет. Например, если вы запускаете приложение в режиме yarn-client, то драйвер будет на машине, с которой вы запускали, а значит драйвера (в отличие от логов воркеров) будут писаться локально на STDOUT.
Многие вещи в Impala появились гораздо раньше, чем в Hive, да и стабильная версия у неё уже 2.1. Скорее всего у вас просто был неудачный опыт или вы привычки к Hive, у которого всё-таки немного другие цели.
Обычно данные просто записываются в HDFS. Это ещё один пример преимущества распределённой файловой системы: данные из каждой партиции запишутся локально и в паралелли. Если же надо получить на выходе ровно один файл, то можно вызвать .repartition(1) и тот же .saveAsHadoopFile() — все данные пойдут на одну машину, но не в виде одной коллекции, а в виде ленивого итератора, что полностью решает проблему с памятью.
AlexSerbul
Спасибо за развернутый комментарий! Приятно общаться с понимающими коллегами.
Конечно, никто не спорит: «лишь Apache Hive можно считать классическим и надежным инструментом для работы с SQL-запросами по распределенным данным».Да, мы часть задач для MR пишем через эту прибамбасу на PHP. Но где нужна скорость — на java. Про Storm я бы поспорил — имхо он больше для Task Parallel задач, не для Data Parallel как MR или Spark.
Согласен полностью! Парадигма Data Parallel Processing и ее детеныш в виде методики свертки функций — MapReduce учат нас, что данные можно разложить по кусочкам и одновременно на каждом кусочке выполнить функцию. А с точки зрения реализации уже, на уровне железа, будет лучше держать данные ближе к процессам, их обрабатывающим. В идеале. На практике вот что получалось:
1) Когда держали ноды с воркерами HDFS с запущенными там воркерами Spark — при интенсивной работе с данными воркеры Spark резко замедляли воркеры HDFS ;-)
2) Настройки ОС, конфигурация железа для нод с HDFS и нод со Spark — сильно отличается. В первом случае — диски большие, во втором — памяти побольше и ядер.
3) Коллеги из Hadoop одно время назад реализовали возможность «привязки» данных к серверным стойкам. Вот это способ видится мне более рациональным — когда в одной стойке под одним маршрутизатором живут машины с HDFS и машины Spark с разной железной конфигурацией.
4) Данных может быть на 2 сервера HDFS, а серверов для их обработки нужно 20. И наоборот. У нас так.
5) В Амазоне что машины с s3, что машины локальные — находятся примерно одинаково далеко :-) Вот у нас и не было выигрыша. Но если бы мы поднялись в изолированном кластере, а там есть такая услуга, то да, было бы быстрее хранить данные в HDFS.
Да, так и делаем. Для NoSQL у нас DynamoDB.
Да смотрел я ее, но она вроде логи только получает не грохая? Логи просмотреть можно и через веб-админку. Нам нужно чтобы логи ротировались и удалялись из кластера. Пока приходится это делать кучей настроек в разных местах и скриптами на bash, лазающими и грохающими что найдут мусорного :-)
Нет проблемы с памятью — проблемы со скоростью выгрузки данных из RDD на диск. Если получать итератор по данным и качать из него — 3 дня ждать приходится :-) Repartition может выполняться неразумно долго, пробовали. Я код накидал выше, он позволяет выкачивать RDD гораздо быстрее в нашем случае.
ffriend
Я всё равно не понял, как Hive может быть ещё и классикой среди распределённых файловых систем, если файловой системой он не является. Ну да ладно, будем считать, что это моё личное восприятие.
Я не очень понимаю, как вы определяете data parallelism и task parallelism. Data parallelism обычно используется в контексте GPU или SIMD инструкций CPU — single instruction, multiple data. Task parallelism обычно ставится им в противоположность как возможнось исполнять в разных потоках разный код. Что Spark, что Storm позволяют одинаково легко делать и то, и другое.
Я ещё раз повторюсь, что смысл map-reduce именно в локальности данных, а не в параллельной обработке. Смотрите: возьмём 20 серверов MySQL, возьмём 20 серверов приложения и будем каждым инстансом приложения обрабатывать только часть данных из MySQL. Нужен здесь map-reduce? Нет, подобная схема существовала задолго до появления оной парадигмы.
Теперь рассмотрим другой пример: есть 20 серверов с индексами веб-страниц и 100 тысяч поисковых запросов в секунду. Если мы поставим отдельно 20 серверов приложений, то на каждый такой запрос сервер приложения должен будет выкачивать несколько гигабайт данных с серверов индексов, сканировать их и выдавать топ-10 результатов. Оверхед на передачу данных фантастический. С другой стороны, если собственно поисковый механизм перенесём на сервера индексов, то передавать придётся не все данные, а только топ-N записей: алгоритм локально подсчитает подходящие результаты (map), а потом передаст только наиболее подходящие на центральный сервер, который закончит работу (reduce). Вот это и было мотивацией создания map-reduce, а не просто распараллеливание работы.
А как вы это замеряли? DataNode-ы требуют минимум CPU и упираются в первую очередь в скорость диска. Spark требует CPU и памяти, но с диском работает через те же DataNode-ы. Возможно, вы получились где-то бонус от кеша HDFS, который тоже в памяти, но это довольно специфичный случай и из него можно получить профит и без, и со Spark-ом.
Так а кто мешает поставить сбалансированные ноды? Благо, даже если вы изначальное не угадали с конфигурацией, то что Amazon, что выделенные стойки с серверами позволяют добавлять ресурсы (CPU, RAM, disk) по необходимости.
В Amazon это называется compute-intensive nodes. Т.е. нет никакой проблемы взять машины с небольшим диском, но большим количеством ядер и памяти.
С другой стороны, если данных у вас действительно настолько мало, что вы можете свободно гонять их по сети, то и map-reduce как таковой вам не нужен. Spark здесь может дать дополнительное удобство, но принципиального отличия от тех же 20 серверов приложений, которые тянут данные с 20 серверов MySQL, нет.
Так вот поэтому map-reduce и размещает ноды с данными на тех же машинах, что и ноды-обработчики — тогда они находятся максимально «близко» ;)
Выше — это в статье? Не могу найти.
AlexSerbul
Пример простой задачи. Нужно сжать, зашифровать и скопировать 100 млн. файлов из хранилища А в хранилище Б. Алгоритмически, время считывания файла из хранилища мизерно, по сравнению со временем сжатия и копирования файла. Задача легко решается на Hadoop/Spark — удобно, если грохается операция по причине временных внутренних ошибок, она перезапускается снова.
Это типичный кейс data parallel задачи. При чем же тут локализация хранения данных и процессов, работающих с ними?
AlexSerbul
Еще пример. Spark по дефолту использует немалую часть ОЗУ для кэширования промежуточных результатов трансформации RDD. Считывание RDD во многих задачах (у нас так) происходит в самом начале. Затем данные трансформируются в памяти. Алгоритмически не очень важно, где данные лежат — в файле на диске, в HDFS, в s3 — это 5% времени выполнения алгоритма.
Я к тому, что не стоит привязываться к концепции хранить данные именно в HDFS рядом с воркерами, их обрабатывающими. Идея хорошая, но Spark пошел гораздо дальше и, кстати, заметьте, он хранит временные данные, которые не помещаются в ОЗУ не в HDFS, а на локальных дисках.
ffriend
Так именно что data parallel, а не map-reduce, о чём я и говорю. Вместо Spark можно взять Storm, чистый Akka Cluster, любую реализацию MPI или просто 20 инстансов приложения, и результат будет точно таким же.
Этот пример ещё лучше: если скачивание данных занимает всего 5% времени, то гораздо проще вычитать их напрямую на те же 20 серверов и крутить ими в памяти как угодно, а не только через абстракцию RDD. Т.е. всё это можно делать, и Spark даёт немало инструментов для этого, но к map-reduce не имеет к этому никакого отношения. А если у нас нет map-reduce, то и список возможных инструментов сильно расширяется. Но это уже совсем другой разговор.
С точки зрения внутреннего устройства, центральное понятие Spark-а — это партиция, т.е. данные, которые хранятся на одной машине и обрабатываются одним процессом (см. метод RDD.compute()). Т.е нам уже известно, где будут обрабатываться данные. Тогда логично, что партицию сохраняют на ту же машину.
AlexSerbul
К сожалению, гоняем одно и тоже из пустого в порожнее :-) Но, думаю, мы понимаем друг друга хорошо.
AlexSerbul
Но стало интересно, откуда так распространилось заблуждение, что MapReduce без Data Locality — мало полезен :-) Снова прочитал исходную статью создателей концепции, коллег из Google. Глянул в википедию английскую — может там написали что вредное, не нашел. Посмотрел в глаза Доугу, глаза серьезные :-)
Везде подчеркивается, что основной смысл парадигмы MapReduce — позволить разработчикам писать в функциональном стиле функи Map и Reduce. А об остальном позаботится платформа. Пожелание к платформе — отказоустойчивость, эффективный shuffle ключей к значениям. И немного лишь написано, что эффективнее будет, если данные будут читаться локально в п. 3.4 папера гугла.
Поэтому я не понимаю, поясните, откуда и где вы взяли идею главенствования локальности данных? :-)
ffriend
Да пожалуйста, открываем Википедию и видим:
Про пункт 3.4 пейпера, который так и называется — Locality — вы уже сами написали. Обратите внимание, что алгоритм всегда пытается размещать воркеров на тех машинах, где есть реплика данных (т.е. вариант, когда они на разных машинах, даже не рассматривается), и только если это невозможно из-за загруженности машин, обработка запускается в другом месте. На практике 95-100% работы выполняется локально.
Если вам мало этого пунтка в пейпере, обратите внимание на схему, где явно указаны read и remote read. Да и если вчитаться, думаю, будет ещё много упоминаний про это.
Зачем читать биографию Дуга Каттинга — не знаю, он не только Хадупом отличился.
Вообще, необходимость локальности становится очевидной, когда начинаете работать с действительно большими данными. В одной из последних задач нам нужно было выкачивать 0.5Tb данных из S3 и обрабатывать их на Spark. Выкачивание и складывание на HDFS занимало час, обработка (вместе с чтением из HDFS) — 40 минут. Вот такая математика.
AlexSerbul
Я похоже понял, где затык у нас :-)
Я отстаиваю точку зрения, что часто не важно, где данные хранятся изначально. Spark будет вначале всасывать их в ОЗУ с s3, с HDFS, с Луны всеми нодами и формировать RDD, чтобы затем прогнать по цепочке, допустим мапперов. Разумеется, конечно, после первого маппера и shuffle — данные станут нафиг другими, могут вырасти в размере в 1000 раз от того, что хранилось первоначально в HDFS и разойдутся по воркерам и каждый, допустим, редьюсер будет работать со своим ключем локально. Если бы он не работал локально — для получения каждого значения пришлось бы ходить по TCP/IP, что смертоубийственно :-) Это мы хорошо с вами понимаем.
Во время выполнения DAG Spark, как сами понимаете, не будет скорее всего уже что-то читать из HDFS — если ОЗУ достаточно, он все уже туда втянет и будет тасовать. Иначе смысл использовать Spark без размещения данных в ОЗУ нод кластера.
Т.е. HDFS, s3, Луна — учитываются в момент всасывания данных, и, по закону Амдала, являются последовательной частью алгоритма, которую параллелить иногда бесполезно.
В нашем случае (сейчас над нашими 150ГБ и вашими 500ГБ смеется конечно Google и Facebook) мы вытягивали из s3 150ГБ минут 10, затем колбасили N часов в кластере Spark. Когда положили данные на ноды со Spark — первая часть стала быстрее, а вторая — медленнее на M часов. Когда положили данные рядом на HDFS ноды, стало по времени как ходить в s3 — минут 10.
Дискуссию считаю полезной, т.к. мы определили 2 понятия локальности данных: начальную и рабочую. Начальная по закону Амдала может на время выполнения задания не влиять, а вторая — влияет и очень сильно.
ffriend
Только вот скорость считывания с HDFS будет ограничена исключительно скоростью работы с диском, а вот S3 будет ограничена и диском, и сетью, которая как бы одна на всех.
Впрочем, убеждать вас в чём-то я не собираюсь: у вас принципиально не те задачи и не те объёмы, для которых важна локальность данных. Появлятся те задачи, сами всё прочувствуете.
Прелесть абстракции RDD как раз в том, что это ленивая коллекция данных, т.е. данные будут читаться с диска только по мере надобности. Это тоже чертовски важный аспект, потому что если у вас 20Тб данных на диске, но вам нужно найти только 10 записей, то в памяти в любой момент времени будет не больше этих самых 10 записей.
На ноды со Spark — это просто на локальный диск? Если да, то скорее всего вы не дали шедулеру самому распределить работу между нодами, вот и получилось, что некоторые справлялись быстрее нужного, а другие отставали и тормозили весь процесс.
500Гб — это ежедневный прирост свежих данных. А алгоритмы могут гоняться и за 1, и за 7 и за 30 дней. Добавьте к этому, что RAM-а у нас всего 2Tb, и становится совсем не смешно. Вот в такой ситуации и приходится думать про локальность, избирательное кеширование и прочие оптимизации.
ffriend
Ну и да, попробуйте всё-таки придумать, чем Spark без локальности данных может быть лучше чем, скажем, голый Akka Cluster.
AlexSerbul
Да ничем не лучше, все ляжет к черту. С вашего позволения только поднимусь на уровень выше к модели акторов. Они же тупые, акторы, читают сообщения, выполняют и отвечают результатами. Но! Если хранить очередь сообщений на той же машине, где и актор-получатель (только не в HDFS, а локально на диске, что Spark и делает, сохраняя не помещающиеся в памяти данные на диск — тормозя конечно при этом все и вся) — все станет подобно Spark.
ffriend
Да почему же, ваша задача со сжатием 100 миллионов файлов и перекладыванием в другое место прекрасно решается через Akka Cluster, причём памяти ему нужно будет по максимальному размеру одного файла.