По материалам лекции проф. David J. DeWitt на ежегодном собрании пользователей SQL Server Что такое Большие Данные (Big Data) и как их образатывать?

Обычно размещаются на больших кластерах, построенных на дешевых процессорах. Facebook (в 2009 г.) имел кластер из 2700 узлов с объемом дисковой памяти 60PB ( 100PB в 2012)!! Что означает термин Большие Данные? 2 ? Собрание огромного количества записей – десятки и сотни PBs Для кого-то это использование т.н.NoSQL систем или параллельных реляционных БД

1 zettabyte? = 1 million petabytes = 1 billion terabytes = 1 trillion gigabytes 3 Sources: "Big Data: The Next Frontier for Innovation, Competition and Productivity." US Bureau of Labor Statistics | McKinsley Global Institute Analysis Petabytes Марс Земля 35ZB = достаточно данных, чтобы образовать стопку CD дисков (0.8GB каждый) высотой с половину пути от Земли до Марса, или 70 стопок от Земли до Луны 35ZB = достаточно данных, чтобы образовать стопку CD дисков (0.8GB каждый) высотой с половину пути от Земли до Марса, или 70 стопок от Земли до Луны Если любите аналогии…

4 Почему внезапный взрыв интереса? Значительно возросшее количество и разнообразие источников, вырабатывающих огромный объем данных Аналогово-цифровые датчики(координатные, звуковые, …) Web 2.0 текст (твитер, википедия, … ) Web клик-поток Понимание, что некоторые данные слишком ценны, чтобы удалять их, либо само удаление слишком дорого. Существенное снижение стоимости оборудования, особенно – дисковой памяти Если бы цена была $500/GB (в 1994 г.), не было бы революции Больших Данных (в 2014 г. $0.05/GB)

Как работать с Большими Данными? 5 Использовать NoSQL Facebook - 20PB на 2700 узлах Bing – 150PB на узлах Ebay – 5PB на 532 узлах Использовать параллельную (реляционную) БД eBay – 10PB на 256 узлах

6 Что означает NoSQL? Не то, что SQL не должен использоваться для образотки Больших данных. Не то, что SQL и реляционные БД устарели и отмирают. Не только SQL. Для определенного круга задач другие подходы к хранению и образотке информации могут быть более эффективными, особенно с точки зрения легкости расширения (масштабируемости) информационной базы при постоянном добавлении новых данных. Не путать с NewSQL – новый подход к построению реляционных БД для приложений, ориентированных на образотку большого количества транзакций, сочетающий удобства SQL с масштабируемостью NoSQL систем.

7 Почему NoSQL? Более гибкая модель данных. Не требуется заранее схема. Может использоваться популярный JSON (JavaScript Object Notation) формат для описания данных. Ослабление требования целостности/согласованности данных на условие конечной (возможной в будущем) согласованности. Готовность поступиться целостностью ради доступности. Дешевое/бесплатное ПО. Не надо знать языки кроме стандартных Java/C/C++ (предположительно :-) ) освоенных в школе/университете. Более быстрый доступ к данным.

Более быстрый доступ к данным SQL: 8 NoSQL: Не требуется очистка! Не требуется преобразование! Не требуется загрузка! Анализ там где данные оказались! СУБД Данные поступают Разработка схемы БД Очистка данных Преобразование данных Загрузка данных SQL запросы Иногда называют Сначала модель Иногда называют Модель позже Данные поступают Приложение 1 2 NoSQL система

Два основных типа NоSQL систем Хранилище ключ(и): значение Примеры: MongoDB, CouchBase, HBase, Cassandra, Windows Azure, … Гибкая модель данных типа JSON Записи распределены между узлами кластера путем хеширования значения ключа (sharding) Поиск/извлечение/обновление отдельной записи по ключу Hadoop Программная среда для создания (автоматически) масштабируемых и отказо-устойчивых систем сбора, хранения и образотки больших объемов данных. Модель данных обычно отсутствует Записи хранятся в распределенной файловой системе. 9 Походе на систему образ. транзакций без SQL Похоже на склад данных (data warehouse) без SQL

Два мира – новая реальность Структурированный Не структурированный & Реляционные СУБД Структурированные данные с (заранее) известной схемой ACID (АСИН) Транзакции Использование SQL в качестве языка Модель жесткой согласованности ETL(извлечение,преобразование,загрузка) Дольше до получения какого-то результата Зрелость, стабильность, эффективность NoSQL Системы (Не-)(полу-)структурированные данные, схема (заранее) не известна ACID не гарантируется Транзакции не поддерживаются SQL не используется Возможная согласованность ETL не требуется Конкретный результат м.б. быстрее Гибкость 10

Что нас ожидает? Мир действительно изменился Реляционные СУБД больше не являются основным и подавляющим инструментом хранения и образотки данных Новые подходы должны приниматься как новая реальность, требуется понимание как наилучшим образом использовать новые технологии, такие как Hadoop Реляционные СУБД продолжат доминировать в системах образотки транзакций и в небольших до средних размеров хранилищ данных Много пользователей будут вынуждены иметь дело с обоими мирами 11

Требования: Требования: Масштабируемость до PB и тысяч узлов Масштабируемость до PB и тысяч узлов Высокий уровень отказоустойчивости Высокий уровень отказоустойчивости (Относительная) простота программирования (Относительная) простота программирования 12 Огромный объем данных из клик-потоков, которые необходимо сохранять и анализировать 2006 Hadoop GFS + MapReduce Распределенная и отказо-устойчивая Распределенная и отказо-устойчивая новая парадигма программирования новая парадигма программирования Сохранение Обработка 2003 MR/GFS Hadoop = HDFS + MapReduce Сохранение Обработка

Возможность использования для «Хранилищ Данных» Масштабируемость и высокий уровень отказоустойчивости Возможность быстро анализировать большие множества записей без необходимости начального построения модели, очистки и загрузки данных Низкая начальная стоимость оборудования и программного обеспечения Простая парадигма программирования для написания и выполнения программ (автоматически) масштабируемых до тысяч узлов и PB данных

HDFS – распределенная, отказо-устойчивая файловая система MapReduce – набор средств для разработки/выполнения распределенных отказо-устойчивых приложений Hive & Pig – языки разработки приложений(с элементами SQL) Sqoop – пакет программ для передачи данных между HDFS и реляционной БД А также другие… 14 HDFS Map/Reduce Hive & Pig Sqoop HBase Средства загрузки Отчеты Рел.БД

15 HDFS Map/Reduce Hive & Pig Sqoop Реляционная БД 5

Основа всей среды разработки Hadoop Цели HDFS: Масштабируемость до тысяч узлов Предположение, что отказы (оборудования и ПО) возникают достаточно часто Предназначена для хранения небольшого количества очень больших файлов Файл записывается один раз, а читается многократно Традиционная иерархическая организация файлов и оглавлений Высокая степень переносимости между разными платформами 16 HDFS Map/Reduce Hive & Pig Sqoop

17 Большой файл … Большой файл … 6440MB Block 1 Block 1 Block 2 Block 2 Block 3 Block 3 Block 4 Block 4 Block 5 Block 5 Block 6 Block 6 Block 100 Block 100 Block 101 Block MB … 40MB Блок 1 Блок 1 Блок 2 Блок 2 Обозначим блоки разным цветом Блок 3 Блок 3 Блок 4 Блок 4 Блок 5 Блок 5 Блок 6 Блок 6 Блок 100 Блок 100 Блок 101 Блок 101 т.е. Размер блока = 64MB Файлы HDFS состоят из множества блоков Обычный размер блока 64MB Каждый блок хранится как отдельный файл в локальной файловой системе (например, NTFS)

Стратегия размещение по умолчанию: Первая копия записывается на узел, создающий файл ( близость записи ) Вторая копия записывается на узел в той же стойке (чтобы минимизировать сетевой трафик между стойками) Третья копия записывается на узел в другой стойке Узел 5 Узел 4 Узел 3 Узел 2 Узел 1 18 Блок 1 Блок 1 Блок 3 Блок 3 Блок 2 Блок 2 Блок 1 Блок 1 Блок 3 Блок 3 Блок 2 Блок 2 Блок 3 Блок 3 Блок 2 Блок 2 Блок 1 Блок 1 Фактор дублирования = 3 Цели: равномерная загрузка, быстрый доступ, отказо-устойчивость

Узел [обслуживания] имен (NameNode) – один на кластер Отвечает за операции над метаданными (данные о данных), избыточность/дублирование данных, местоположение блоков Резервный узел имен(Backup Node) – резервная копия узла имен Узлы данных(DataNodes) – один на каждый узел кластера Отвечает за хранение блоков файла Предоставляет данные файла в ответ на запрос приложения 19 Узел имен Мастер Резервный узел имен (копия) DataNode Узел данных Робот

20 Узел имен Резервный узел имен Резервный узел имен Узел данных (проверка состояния, балансирование загрузки, дублирование блоков, и т.д.) Узлы записывают на свои локальные диски

Узел данных 21 Узел имен Резервный Узел имен Резервный Узел имен Большой файл Большой файл HDFSКлиентHDFSКлиент {узел 1, узел 2, узел 3} основываясь на факторе дублирования (по умолчанию - 3) Клиент передает блоки указанным узлам напрямую Клиент передает блоки указанным узлам напрямую {узел 2, узел 4, узел 5} {узел 1, узел 3, узел 5} {узел 2, узел 3, узел 4} И так далее… Узел имен указывает клиенту, куда записать каждый блок файла

22 Узел имен Резервный узел имен Резервный узел имен Большой файл Большой файл HDFSКлиентHDFSКлиент Возвращает расположение блоков файла Возвращает расположение блоков файла Узел данных Поток блоков от узлов ранних

Типы отказов: Ошибки или полный отказ дисков Отказы узлов данных Отказы стоек/сетевых коммутаторов Отказы узла имен Полный отказ дата-центра HDFS разрабатывалась, предполагая частые отказы (как оборудования так и ПО) 23 NameNode DataNode

Узел данных 24 Узел имен Резервный узел имен Резервный узел имен Узел данных Узел имен обнаруживает отказ узла данных Каждый блок данных автоматически дублируется на оставшихся узлах удовлетворяя «фактору дублирования» Каждый блок данных автоматически дублируется на оставшихся узлах удовлетворяя «фактору дублирования» Узел данных

25 Узел имен Резервный узел имен Резервный узел имен Узел данных Отказ не катастрофический, так как имеется резервный узел Отказ не катастрофический, так как имеется резервный узел Восстановление может быть автоматическим или требовать ручного вмешательства в некоторых случаях Восстановление может быть автоматическим или требовать ручного вмешательства в некоторых случаях

26 Узел имен Резервный узел имен Резервный узел имен Узел данных Узел имен обнаруживает, что новый узел данных добавлен к кластеру Узел имен обнаруживает, что новый узел данных добавлен к кластеру Узел данных Блоки данных равномерно перераспределяются Блоки данных равномерно перераспределяются Узел данных

Высокая степень масштабируемости Тысячи узлов данных и огромные (сотни TB) файлы Большой размер блоков для повышения скорости последовательного ввода/вывода Не используются «зеркальные диски» и RAID. Меньшая цена Использвание единого инструмента (тройное дублирование блоков) для борьбы с различными типами отказов вместо различных механизмов Недостатки Размещение блоков «прозрачно» для ПО верхнего уровня Многие возможности оптимизации (с успехом используемые в параллельных реляционных БД) не применимы Принцип «одна запись – многократное чтение» может быть слишком ограничительным 27 Почему?

28 HDFS Map/Reduce Hive & Pig Sqoop Relational Databases

Среда программирования (библиотеки и исполняющая система) для анализа данных, хранимых в HDFS MapReduce задание состоит из двух функций/этапов: Пользователь только разрабатывает Map и Reduce функции MR среда «собирает» все вместе и координирует выполнение Map и Reduce функций на кластере, обеспечивая отказо-устойчивость масштабируемость 29 map() reduce() разделяй & властвуй (функция отображения ) разделяй & властвуй (функция отображения ) объединяй & сокращай (функция агрегирования) объединяй & сокращай (функция агрегирования)

В основном, это… 1. Взять большую задачу и разделить ее на под-задачи 2. Выполнить (одну и ту же) функцию отображения для всех под- задач 3. Соединить результат всех под-задач (функция агрегирования) 30 DoWork() … … … Вывод MAP REDUCE

31 … … … … Получение и вывод результата Reducer Reducer Reducer сорти- ровка и груп- пиро- вание по ключу сорти- ровка и груп- пиро- вание по ключу Узел данных MapperMapper MapperMapper MapperMapper MapperMapper

Опер Задач сохраняет промежуточные результаты Промежуточные данные сохраняются в локальной файловой системе Уровень MapReduce Уровень HDFS Узел имен Hadoop hadoop- Узел Данных 1 hadoop- Узел Данных 2 hadoop- Узел Данных 3 hadoop- Узел Данных 4 32 Оператор Работ Оператор Работ Оператор Задач Оператор Задач Оператор Задач Оператор Задач Оператор Задач Оператор Задач Оператор Задач Оператор Задач Оператор Работ контролирует и следит за состоянием Операторов Задач Мастер Роботы - Координирование всех MR задач и событий - Управление очередями работ и графиком выполнения - Поддержка и контроль Операторов Задач - Перемещение/возобновление MR задач, если нужно - Использование «контрольных точек» для восстановления после отказа Выполнение индивидуальных MR задач, назначенных Оператором Работ (в отдельной JVM) Узел Данных Узел имен Уровень MapReduce Уровень HDFS

33 Оператор Работ Оператор Работ Опер Задач локальные файловые системы узлов данных Map функции назначаются Опер.Задач на каждом Узле Данных Map функции назначаются Опер.Задач на каждом Узле Данных Поставляет работы Оператору Работ MRклиентMRклиент Работы ставятся в очередь Работы ставятся в очередь MapperMapper Mapper Mapper Функции отображения запускаются на отдельной JVM и выполняются Функции отображения сохраняют промежуточные результаты Начинается этап объединения/сокращения Reducer

Задачи отображения $ $ $ $ $ $ $ $ $ $ $ $ $ $ $ $ $ $ $ $ $ $60 MapperMapper MapperMapper $ $ $ $ $ $ $ $ $ $ $ $ $ $95 Получить сумму продаж по каждому индексу Узел Данных 3 Узел Данных 2 Узел Данных 1 Блоки файла продаж в HDFS Group By Group By (клиент, индекс, объем продаж) Один сегмент данных на задачу агрегир.

Reducer Задачи агрегирования Reducer 53705$ $ $ $ $ $60 MapperMapper 53705$ $ $ $ $ $25 MapperMapper 53705$ $ $ $ $ $ $ $ $ $ $ $15 Сорти- ровать Сорти-ровать ровать 53705$ $ $ $ $ $ $ $ $ $ $ $15 Агреги- ровать Агреги- ровать Агреги- ровать 10025$ $ $ $ $30 Работа выполнена! Перетасование

36

37 Файл AФайл B Различные ключи HDFS хранит блоки данных (Дублирование не показано) Mapper M Mapper 2 Mapper 1 Mapper 3 Задача отображения генерирует множество пар (ключ, запись) Каждая задача отображения образатывает по одному блоку в раз Reducer 1 Reducer 2 Reducer N Задачи агрегирования выполняют само соединение Перетасовка и сортировка Перетасовка по сети и сортировка

Действительное число задач отображения M обычно выбирается значительно большим, чем число узлов. Почему? Помогает иметь дело с неравномерность распределения значений ключей (data skew) и отказами Пример: Скажем, M = 10,000 и W = 100 (W - число Map роботов) Каждый робот выполнит (10,000 / 100) = 100 задач отображения Если на этапе отображения возникант неравномерность или отказ, невыполненная задача может быть легко перераспределена для выполнения на другом роботе Проблема неравномерности на уровне агрегирования не исчезает Пример: В запросе получить сумму продаж по каждому индексу, некоторые из индексов могут иметь гораздо больше записей для образотки, чем другие 38

Подобно HDFS, среда разработки MapReduce построена, чтобы обеспечть высокий уровень отказоустойчивости Отказ робота (на этапе отображения или агрегирования) Обнаруживается периодическими проверками, исходящими от Мастера Незавершенные задачи стартуют сначала на другом узле Если узел отказывает после завершения этапа отображения – вся задачи отображения этого узла переделывается, и все задачи агрегирования уведомляются. Отказ Мастера Если отказывает Мастер (-процесс), то вся работа полностью переделывается 39

Весьма высокая отказо- устойчивость Относительно простое программирование распределенных вычислений для типичных функций типа агрегирования Среда MR позволяет программисту не заботиться об отказах и сбоях 40 Схема/структура данных кодируется в каждом приложении Недостаток общей схемы данных Делает совместное использование данных между приложениями весьма затруднительным Невозможно использование аппарата реляционых БД (таких как индексы, ограничения целостности, представления, …) для эффективной образотки Нет декларативного языка для описания и доступа к данным типа SQL

41 HDFS Map/Reduce Hive & Pig Sqoop Relational Databases

и использовали разные подходы к использованию декларативных языков для работы с данными (подобно SQL) в дополнение к MapReduce 42 HIVE Facebook разработала язык типа SQL, названный HIVE PIG Yahoo разработала более процедурный язык, названный PIG Оба подхода используют MapReduce как язык описания/выполнения работ Запросы/программы на языках Hive и Pig компилируются в последовательность работ MapReduce

Рассмотрим два файла (множества записей): 43 Rankings ( pageURL STRING, pageRank INT, avgDuration INT ); UserVisits ( sourceIP STRING, destURL STRING visitDate DATE, adRevenue FLOAT,.. ); Запрос: Найти sourceIP адрес, генерирующий наибольший доход, вместе со средним рангом посещенных страниц соединение

package edu.brown.cs.mapreduce.benchmarks; import java.util.*; import org.apache.hadoop.conf.*; import org.apache.hadoop.io.*; import org.apache.hadoop.mapred.*; import org.apache.hadoop.util.*; import org.apache.hadoop.mapred.lib.*; import org.apache.hadoop.fs.*; import edu.brown.cs.mapreduce.BenchmarkBase; public class Benchmark3 extends Configured implements Tool { public static String getTypeString(int type) { if (type == 1) { return ("UserVisits"); } else if (type == 2) { return ("Rankings"); } return ("INVALID"); } /* (non-Javadoc) org.apache.hadoop.util.Tool#run(java.lang.String[]) */ public int run(String[] args) throws Exception { BenchmarkBase base = new BenchmarkBase(this.getConf(), this.getClass(), args); Date startTime = new Date(); System.out.println("Job started: " + startTime); 1 // Phase #1 // JobConf p1_job = base.getJobConf(); p1_job.setJobName(p1_job.getJobName() + ".Phase1"); Path p1_output = new Path(base.getOutputPath().toString() + "/phase1"); FileOutputFormat.setOutputPath(p1_job, p1_output); // // Make sure we have our properties // String required[] = { BenchmarkBase.PROPERTY_START_DATE, BenchmarkBase.PROPERTY_STOP_DATE }; for (String req : required) { if (!base.getOptions().containsKey(req)) { System.err.println("ERROR: The property '" + req + "' is not set"); System.exit(1); } } // FOR p1_job.setInputFormat(base.getSequenceFile() ? SequenceFileInputFormat.class : KeyValueTextInputFormat.class); if (base.getSequenceFile()) p1_job.setOutputFormat(SequenceFileOutputFormat.class); p1_job.setOutputKeyClass(Text.class); p1_job.setOutputValueClass(Text.class); p1_job.setMapperClass(base.getTupleData() ? edu.brown.cs.mapreduce.benchmarks.benchmark3.phase1.TupleWritableMap.class : edu.brown.cs.mapreduce.benchmarks.benchmark3.phase1.TextMap.class); p1_job.setReducerClass(base.getTupleData() ? edu.brown.cs.mapreduce.benchmarks.benchmark3.phase1.TupleWritableReduce.class : edu.brown.cs.mapreduce.benchmarks.benchmark3.phase1.TextReduce.class); p1_job.setCompressMapOutput(base.getCompress()); 2 // Phase #2 // JobConf p2_job = base.getJobConf(); p2_job.setJobName(p2_job.getJobName() + ".Phase2"); p2_job.setInputFormat(base.getSequenceFile() ? SequenceFileInputFormat.class : KeyValueTextInputFormat.class); if (base.getSequenceFile()) p2_job.setOutputFormat(SequenceFileOutputFormat.class); p2_job.setOutputKeyClass(Text.class); p2_job.setOutputValueClass(Text.class); p2_job.setMapperClass(IdentityMapper.class); p2_job.setReducerClass(base.getTupleData() ? edu.brown.cs.mapreduce.benchmarks.benchmark3.phase2.TupleWritableReduce.class : edu.brown.cs.mapreduce.benchmarks.benchmark3.phase2.TextReduce.class); p2_job.setCompressMapOutput(base.getCompress()); // Phase #3 // JobConf p3_job = base.getJobConf(); p3_job.setJobName(p3_job.getJobName() + ".Phase3"); p3_job.setNumReduceTasks(1); p3_job.setInputFormat(base.getSequenceFile() ? SequenceFileInputFormat.class : KeyValueTextInputFormat.class); p3_job.setOutputKeyClass(Text.class); p3_job.setOutputValueClass(Text.class); //p3_job.setMapperClass(Phase3Map.class); p3_job.setMapperClass(IdentityMapper.class); p3_job.setReducerClass(base.getTupleData() ? edu.brown.cs.mapreduce.benchmarks.benchmark3.phase3.TupleWritableReduce.class : edu.brown.cs.mapreduce.benchmarks.benchmark3.phase3.TextReduce.class); 3 // // Execute #1 // base.runJob(p1_job); // // Execute #2 // Path p2_output = new Path(base.getOutputPath().toString() + "/phase2"); FileOutputFormat.setOutputPath(p2_job, p2_output); FileInputFormat.setInputPaths(p2_job, p1_output); base.runJob(p2_job); // // Execute #3 // Path p3_output = new Path(base.getOutputPath().toString() + "/phase3"); FileOutputFormat.setOutputPath(p3_job, p3_output); FileInputFormat.setInputPaths(p3_job, p2_output); base.runJob(p3_job); // There does need to be a combine if (base.getCombine()) base.runCombine(); return 0; } 4 44 Даже не пытайтесь это читать! Запрос требует 3 MapReduce работы, так как написание программы соединения в MR – довольно трудоемкая задача 4 страницы мелким шрифтом!!!

Только пол-страницы крупным шрифтом! SELECT sourceIP, totalRevenue, avgPageRank FROM SELECT sourceIP, sum(adRevenue) as totalRevenue, avg(pageRank)as avgPageRank FROM Rankings as R, Uservisits as UV WHERE R.pageURL = UV.destURL and UV.visitDate between Date ( ) and Date ( ) GROUP BY UV.sourceIP ORDER BY totalRevenue DESC limit 1; 45 В Facebook сделали вывод: … MapReduce был не простым для пользователей … … пользователи тратили несколько часов (или даже дней) для написания MR программ даже для очень простого анализа Из работ выполняемых ежедневно в Facebook, только 500 – работы написанные непосредственно в MapReduce! Остальные- в HiveQL!! Да здравствует SQL!!!

46 Запросы на декларативном языке(типа SQL) Большие объемы данных (использование Hadoop) HiveQL = Удобные средства написания запросов(SQL) + MapReduce HIVE Отказо-устойчивая система

Подобно реляционным БД, данные хранятся в таблицах Типы данных богаче, чем в SQL Простые типы: int, float, string, date Сложные типы: ассоциативные массивы, списки, структуры Example: 47 CREATE Table Employees ( Name string, Salary integer, Children List > )

Как и в параллельных реляционных БД, Hive таблицы могут храниться в нескольких разделах Пример: Sales(custID, zipCode, date, amount) partitioned by state 48 Alabama Alaska Wyoming … Sales HDFS оглавление 1 HDFS файл на каждый штат Create Table Sales( custID INT, zipCode STRING, date DATE, amount FLOAT) Partitioned By (state STRING) Hive DDL: Записи в файле могу не содержать название штата! Каждый HDFS файл в свою очередь разделен на большие блоки, которые продублированы и распределены по узлам кластера

49 SELECT zipCode, sum(amount) FROM Sales WHERE getDate()-30 < date < getDate() GROUP BY zipCode Alabama Alaska Wyoming … Sales HDFS оглавление HDFS файлы Sales(custID, zipCode, date, amount) partitioned by state Запрос 1: За последние 30 дней выдать объем продаж по каждому индексу (zipCode): Запрос будет быполнен по всем штатам/разделам

50 Alabama Alaska Wyoming … Sales HDFS оглавление HDFS файлы Sales(custID, zipCode, date, amount) partitioned by state SELECT zipCode, sum(amount) FROM Sales WHERE State = Alabama and getDate()-30 < date < getDate() GROUP BY zipCode Запрос 2: За последние 30 дней выдать объем продаж по каждому индексу (zipCode) штата Alabama: Запрос будет выполнен по всем разделам (одного узла), содержащим записи штата Alabama

Оптимизация запросов Ограниченная статистика (только размер файлов) делает практически невозможной оптимизацию запросов на основе стоимости Используются простые эвристические приемы для выборки перед соединением, раннего исключение столбцов,... Результат частичной оптимизации – направленный граф (без циклов) MapReduce работ на языке Java 51 Блокоптимизации Запрос выполняется (контролируется) стандартным MR планировщиком Отказо-усточивость обеспечивается автоматически Может пережить отказы узла/диска/сетевого маршрутизатора в середине образотки запроса Планы выполнения выглядят сложнее поскольку каждый шаг образотки использует в качестве входа отдельный файл HDFS Эвристические приемы Планы сложные для понимания и анализа

Оборудование Кластер из 9 HP узлов, каждый имеет 2 4-х ядерных процессорв, 16GB памяти, 4 жестких диска для данных ПО SQL Server PDW (Parallel Data Warehouse) 1 контролирующий узел, 8 рабочих Windows Hadoop Version , Hive Version узел имен, 8 узлов данных Windows Server 2008 Тест таблицы из TPC-H (SF 800) lineitem: 612GB, 4.8B записей orders: 140GB, 1.2B записей 52

Query 1: SELECT count(*) FROM lineitem 53 Query 2: SELECT max(l_quantity) FROM lineitem WHERE l_orderkey > 1000 and l_orderkey < GROUP BY l_linestatus Secs.

2 случая для PDW : i) PDW-U: orders partitioned on c_custkey lineitem partitioned on l_partkey ii) PDW-P: orders partitioned on o_orderkey lineitem partitioned on l_orderkey 54 Демонстрирует преимущество параллельных реляционных БД, в которых способ разделение таблиц позволяет минимизировать перемещение данных между узлами кластера (соединяемые разделы находятся на том же узле) Разделение Hive таблиц по столбцу соединения не дает никаких выгод, поскольку нет средств контроля, куда HDFS размещает блоки данных таблиц Query 3: SELECT max(l_orderkey) FROM orders, lineitem where l_orderkey = o_orderkey

Рассмотрим два файла (множества записей): 55 Rankings ( pageURL STRING, pageRank INT, avgDuration INT ); UserVisits ( sourceIP STRING, destURL STRING visitDate DATE, adRevenue FLOAT,.. ); Запрос: Найти sourceIP адрес, генерирующий наибольший доход, вместе со средним рангом посещенных страниц соединение

56 UV = load userVisits as (sourceIP, destURL, visitDate, adRevenue); UVFiltered = filter UV by visitDate >= and visitDate

57 HDFS Map/Reduce Hive & Pig Sqoop Relational Databases

Все больше данных изначально попадает в неструктурированный мир MapReduce – прекрасное средство для ETL (выбор, очистка, преобразование, загрузка) 58 Sqoop предоставляет утилиту загрузки, выполняемую из командной строки Sqoop (причина 1) Структурированные данные (рел.БД) Неструктурированные данные (Hadoop)

Некоторые задачи анализа проще делать в процедурном языке или в языке типа HiveQL с элементами MapReduce, когда необходимо 59 Sqoop предоставляет: утилиту выгрузки, выполняемую из командной строки Возможность для Map задач доставать данные из реляционной БД, используя SQL Как мы увидим эффективность этого невысока Sqoop (причина 2) Структурированные данные (рел.БД) Неструктурированные данные (Hadoop)

Некоторые приложение требуют данные из обоих миров 60 Только в неструктурированном мире т.к. неструктурированные данные не могут быть загружены в реляционную БД Использвать утилиту выгрузки и доступ к БД для Map задач По-прежнему, эффективность этого невысока Sqoop (причина 3) Структурированные данные (рел.БД) Неструктурированные данные (Hadoop)

RDBMS Map 1 Sqoop Map 2 Sqoop Map 3 Sqoop Cnt Q: SELECT a,b,c FROM T WHERE P Map задача требует результат запроса : Шаг (2): Sqoop генерирует отдельный запрос Q для каждой Map задачи: SELECT a,b,c FROM T WHERE P ORDER BY a,b,c Limit L, Offset X Шаг (1): Sqoop выполняет SELECT count(*) FROM T WHERE P для получения Cnt, числа записей в T 61 Шаг (3): Каждая из задач выполняет свой запрос Каждая map() должна выбрать свое (непересекающееся) подмножество записей T Смещение X свое для каждой задачи. Пример, пусть Cnt 100 и используются 3 Map задачи Для Map 1 Для Map 2 Для Map 3 L=33 L=34 X=0 X=33 X=66 Эффективность будет плохая, поскольку таблица T читается 4 раза! В общем случае, с M Map задачами, таблица T читалась бы В общем случае, с M Map задачами, таблица T читалась бы M + 1 раз!!!!!!

Пересылка данных – это прошлый век! Почему бы не создать систему управления данными которая: Может выполнять запросы, используя данные из обоих миров без необходимости их перемещения из одного в другой Имеет выразительную возможность языков типа HiveQL Назовем такую систему Администратор Данных Предприятия АДП (EDM – Enterprise Data Manager). 62 Администратор Данных Предприятия Структурированные данные (Рел.ДБ) Неструктурированные данные (Hadoop)

Первая попытка создать «Администратор Данных Предприятия» Подобно Hive Использует HDFS для «неструктурированных» данных Использует среду MR для образотки запросов для обеспечения масштабируемости и отказо-устойчивости Поддерживает язык запросов типа HiveQL В отличие от Hive Имеет реляционную БД на каждом узле Использует новый подход разделения выполнения запроса между областями структурированных и неструктурированных данных (прозрачно с точки зрения пользователя) 63

64 Таблицы РБД разделяются между узлами путем хеширования «Неструктурированные» данные хранятся в HDFS SQL запросы компилируются в последовательность работ MapReduce Используется подход разделения выполнения запроса, при этом РБД используется как можно больше Рассмотрим соединениетаблиц A и B RDBMS HDFS MapReduce Работа MapReduce Администратор MR Каталоги Синтаксический анализатор Оптимизатор запросов SQL Запрос RDBMS HDFS MapReduce RDBMS HDFS MapReduce RDBMS HDFS MapReduce RDBMS HDFS MapReduce РБДHDFS MapReduce Пример 1: Обе таблицы A & B в РБД Соединение выполняется полностью в РБД Пример 2: A – в РБД, а B - в HDFS. Если таблица B «небольшая», она загружается в РБД, и соединение выполняется в РБД Если таблица A «небольшая», а B «большая», соединение выполняется как работа MapReduce

Без необходимости загрузки Без необходимости загрузки Неструктур. данные Неструктур. данные Лучшая масштабируемость Лучшая масштабируемость Отказо- устойчивость Отказо- устойчивость SQL Server PDW Системы, основанные на Hadoop значительно проще сравнительной производительности Добавить Пытающиеся достичь

Параллельные РБДHadoop Вычислительная модель -Понятие транзакции -Транзакция – единица исполнения -ACID обеспечивается для параллельных транзакций -Понятие работы -Работа – единица исполнения -Управление параллельным исполнением отсутствует Модель данных -Структурированные данные с известной схемой -Режимы Чтение/Запись -Любые данные -(не-)(полу-)структурированные -Режим Только Чтение Конфигурация оборудования -Покупается для специального применения, дороже обычного -«Собирается» пользователем из дешевых компьютеров «широкого потребления» Отказо- устойчивость -Отказы предполагаются редкими -Отказо-устойчивость для отдельных запросов отсутствует -Отказы предполагаются довольно частыми -Простая, но эффективная отказо-устойчивость на уровне запросов Ключевые характеристики -Эффективность, оптимизация, возможность настройки -Масштабируемость, гибкость, отказо-устойчивость

67 Реляционные БД или Hadoop? (каково будущее?) Вопрос не корректен! Реляционные БД и Hadoop созданы для различных целей/задач Только РБД или только Hadoop не будет основным средством для образотки больших данных Решит ли проблему Администратор Данных Предприятия?