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

пара ключ-значение RDD в целомиспользовать Приходитьруководитьполимеризациясчитать Рассчитать。насодин Общийхотеть Первыйпроходитьодиннекоторыйисходный ETL (Извлечение、изменятьизменять、нагрузка)Держатьделать Приходить Волячислов соответствии сизменятьизменятьдляпара ключ-значениеформа。пара ключ-значение RDD поставлять Понятноодиннекоторыйновыйиз Держатьделатьинтерфейс(Сравниватьнравитьсясистемасчитать Каждыйиндивидуальныйпродуктиз Комментарий,Волячислов соответствии ссерединаключтакой жеизточкадляодин Группа,Волядваиндивидуальный Неттакой жеиз RDD Группировка и слияние и т. д.).

1. Мотив

Spark для Включатьпара ключ-значениетипиз RDD Предусмотрены некоторые собственные операции. Эти RDD называется pair RDD 。PairRDD дамного программизсоставляютхотетьбелый,потому чтодляэтоихпоставлять Понятнопараллельный ДержатьделатькаждыйиндивидуальныйключилиПерезапуск на всех узлахпродолжай считать соответствии группа сточкаиз Держатьделатьинтерфейс。примернравиться,pair RDD поставлять reduceByKey() метод, вы можете уменьшить данные, соответствующие каждому ключу отдельно, и join() метод, вы можете объединить два RDD Элементы с одинаковым средним ключом группируются и объединяются в один RDD。

2. Создать пару RDD

Pair RDDСразуда Юаньбелыйдля2-кортеж（tuple）изRDD。существоватьSparkсерединаиметь Различныйнаправление Режим Создать пару РДД, из которых наиболее распространены два:

• Много места для хранения, пара ключ-значениеизчислов соответствии ссетка Режимвстречасуществоватьчитатьчас Возврат напрямую Зависит от Чтопара ключ-значениечислов соответствии композицияиз pair RDD。
• текущий спросхотеть Пучокодининдивидуальныйобычноиз RDD изменятьдля pair RDD час,Можеткнастраиватьиспользовать map() письмочисло Приходитьвыполнить,передачаизписьмочислонуждатьсяхотетьвозвращатьсяпара ключ-значение。

Example（Python）：

listRDD = sc.parallelize([1, 2, 3, 4])
pairRDD = listRDD.map(lambda x: (x, "brown"))

Example(Scala):

val listRDD = sc.parallelize(List(1, 2, 3, 4))
val pairRDD = listRDD.map(x => (x, "brown"))

3. Операция преобразования пары RDD

Парные СДР могут использовать все операции преобразования, доступные в стандартных СДР. В таблицах 4-1 и 4-2 суммированы некоторые операции преобразования парных RDD:

(1) Операция агрегирования

когдачислов соответствии снаборкпара ключ-значениеформа Группаткатьизчасждать,Агрегация с одинаковым ключомиз Юаньбелыйруководитьодиннекоторыйсистемасчитатьдаочень распространенныйиз Держатьделать。Из Предварительная лекцияразвязать Проходить База RDD на fold() 、 combine() 、 reduce() ждатьдействие Держатьделать,pair RDD Вышеупомянутое правилоиметьсоответствующийизпротивключизизменятьизменять Держатьделать。Spark Существует аналогичный набор операций, позволяющий объединять значения с одним и тем же ключом. Эти операции возвращают RDD,потому чтоэтотэтоихдаизменятьизменять Держатьделатьи Нетдадействие Держатьделать。

полимеризация Держатьделатьхозяинхотеть С участием трехиндивидуальныйписьмочисло：reduceByKey(),foldByKey（）,combineByKey（）。

• reduceByKey(): reduceByKey() и reduce() Взаимнокогдадобрыйпохожий;этоих Всеперениматьодининдивидуальныйписьмочисло,иделатьиспользовать Долженписьмочисловерноценитьруководитьобъединитьи。reduceByKey() встречадлячислов соответствии снаборсерединаиз Каждыйиндивидуальныйключруководитьпараллельныйизснижение Держатьделать,Каждыйиндивидуальныйснижение Держатьделатьвстреча Воляключтакой жеизценитьобъединитьирост Приходить。этовстречавозвращатьсяодининдивидуальный Зависит откаждыйключиверноотвечатьключснижениевне Приходитьиз Узелфруктыценить Группастановитьсяизновыйиз RDD。
• foldByKey(): и fold() Взаимнокогдадобрыйпохожий;этоих Вседелатьиспользоватьодининдивидуальныйи RDD иобъединитьиписьмочислосерединаизчислов соответствии с типом тот же, что и из нулевой цены делать для начальной цены. и fold() Такой же, foldByKey() Держатьделать Местоделатьиспользоватьизобъединитьиписьмочисловернонольценитьи Другойодининдивидуальный Юаньбелыйруководитьобъединитьи,Узелфруктывсе ещедля Должен Юаньбелый。(??)
• combineByKey(): это наиболее часто используется на основе агрегации ключей из функции число. Большинство реализаций числа даиспользоватьэто основаны на ключевом агрегате функции числа. и aggregate() Такой же, combineByKey() Можно использовать возврат пользователя и ввести число соответствии тип размера Нет такой же, как и из возврата ценить. хотеть причина развязать combineByKey() , хотеть Первый причина развязать это существовать в причина число в соответствии с час да нравиться что в причина Каждый индивидуальный Юань белый из。 Зависит от ВcombineByKey() встречатраверсточкаокругсерединаиз Местоиметь Юаньбелый,потому чтоэтот Каждыйиндивидуальный Юаньбелыйизключхотеть Что?возвращатьсябезиметь Встретился,хотеть Что?Сразуи Извпередизопределенныйиндивидуальный Юаньбелыйизключтакой же。
  • нравитьсяфруктыэтотдаодининдивидуальныйновыйиз Юаньбелый, combineByKey() встречаделатьиспользоватьодининдивидуальный Вызовделать createCombiner() изписьмочисло Приходитьсоздай этоиндивидуальныйключверноотвечатьизаккумуляторизисходныйценить。нуждатьсяхотеть Уведомлениеизда,этотодинпроцессвстречасуществоватьКаждыйиндивидуальныйточкаокругсерединаНет.одинявлениякаждыйиндивидуальныйключчаспроисходить,и Нетдасуществоватьвсеиндивидуальный RDD середина Нет.одинявленияодининдивидуальныйключчаспроисходить。
  • нравитьсяфруктыэтотдаодининдивидуальныйсуществоватьвпричинакогдавпередточкаокруг ИзвпередУже встречал изключ,этовстречаделатьиспользоватьmergeValue() направление Закон будет Долженвключить аккумулятор, соответствующий текущей ценеи этому индивидуальному новому изценить.
  • Зависит от ВКаждыйиндивидуальныйточкаокруг Вседанезависимыйвпричинаиз,потому чтоэтотверно Втакой жеодининдивидуальныйключ Можеткиметьмногоиндивидуальныйаккумулятор。нравитьсяфруктыиметьдваиндивидуальныйили ВОЗДажемногоизточкаокруг Всеиметьверноотвечатьтакой жеодининдивидуальныйключизаккумулятор,Сразунуждатьсяхотетьделатьиспользоватьиспользоватьсемьяпоставлятьиз mergeCombiners() направление ЗаконВолякаждыйиндивидуальныйточкаокругиз Узелфруктыруководитьобъединитьи。

Example1: существовать Python используется в reduceByKey() и mapValues() считать Рассчитать Каждыйиндивидуальныйключверноотвечатьизсреднийценить:

rdd.mapValues(lambda x: (x, 1)).reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))

существовать Scala используется в reduceByKey() и mapValues() считать Рассчитать Каждыйиндивидуальныйключверноотвечатьизсреднийценить:

rdd.mapValues(x => (x, 1)).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))

Схема потока данных:

Уведомление： привычный MapReduce серединаизобъединитель(combiner)концепцияизчитатель Можетспособныйуже Уведомлениеприезжать,настраиватьиспользовать reduceByKey() и foldByKey() встреча существовать для Каждый индивидуальный ключ считать Рассчитать Полный бюро из общий Узел фрукты Из вперед Первыйавтоматическийсуществовать Каждый Рабочий столустройствоначальстворуководить本земляобъединитьи。использоватьсемья Нетнуждатьсяхотетьобозначениеобъединитель。более общийизменятьизcombineByKey() Интерфейс Можетк позволяет вам объединять строки для определения.

Example2: combineByKey() иметьмногоиндивидуальныйженьшеньчислоточка Неверноотвечатьполимеризация Держатьделатьизкаждыйиндивидуальныйэтап,потому чтоиочень подходитиспользовать Приходитьразвязатьвыпускатьполимеризация Держатьделатькаждыйиндивидуальныйэтапиздостижениеспособныйрядточка。для Понятнолучшая презентацияcombineByKey() данравитьсячтоработаделатьиз,под Приходитьвзглянинравитьсячтосчитать Рассчитатькаждыйключверноотвечатьизсреднийценить:

существовать Python используется в combineByKey() просить Каждыйиндивидуальныйключверноотвечатьизсреднийценить:

sumCount = nums.combineByKey((lambda x: (x,1)),
    (lambda x, y: (x[0] + y, x[1] + 1)),
    (lambda x, y: (x[0] + y[0], x[1] + y[1])))
sumCount.map(lambda key, xy: (key, xy[0]/xy[1])).collectAsMap()

существовать Scala используется в combineByKey() просить Каждыйиндивидуальныйключверноотвечатьизсреднийценить:

val result = input.combineByKey(
    (v) => (v, 1),
    (acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1),
    (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
    ).map{ case (key, value) => (key, value._1 / value._2.toFloat) }
result.collectAsMap().map(println(_))

Схема потока данных:

(2) Настройка параллельности и количества разделов

Каждыйиндивидуальный RDD Всеиметьзафиксированныйчислоглазизточкаокруг,точкаокругчисло Решать Понятносуществовать RDD выполнить на Держатьделатьчасизпараллельныйстепень。Spark Всегда пробуйте кореньв соответствии предполагается размер кластера один индивидуальныйиметь, что означает по умолчанию ценить, но дайметьчас ты Можетспос обныйхотеть обеспечивает лучшую производительность при параллелизме.

нравитьсячтонастраивать Фестивальточкаокругчисло（параллельныйстепень）Шерстяная ткань？

1. существоватьВыполнить агрегацию или группу «Держатьделать»час,Можеткхотетьпросить Spark делатьиспользоватьданныйизточкаокругчисло。полимеризацияточка Группа Держатьделатьсередина,большоймногочисло Держатьделатьсимвол Всеспособныйперенимать Нет.дваиндивидуальныйженьшеньчисло,этотиндивидуальныйженьшеньчислоиспользовать Приходитьобозначениеточка Группа Узелфруктыилиполимеризация УзелфруктыизRDD източкаокругчисло。нравиться：rdd.reduceByKey((x, y) => x + y, 10)。
2. существоватьудалятьточка Группа Держатьделатьиполимеризация Держатьделать Изснаружииз Держатьделатьсерединатакжеспособныйизменять Изменять RDD източкаокруг。Spark поставлять Понятно repartition() письмочисло。этовстреча Пучокчислов соответствии спроходитьсетьруководитьперетасовать,и создавать новые коллекции кругов източка. Помнить,Восстановить число в соответствии с сиз Держатьделать относительно дорого.。Spark серединатакжеиметь один индивидуальный отличный изменять версия из repartition() , Вызов делать coalesce() 。 ты Может к делать использовать Java или Scala середина из rdd.partitions.size() к И Python серединаиз rdd.getNumPartitions Проверять RDD източкаокругчисло,иправильный Сохранятьнастраиватьиспользовать coalesce() час장 RDD объединитьиприезжать Сравниватьсейчассуществоватьизточкаокругчисломеньшеизточкаокругсередина。

(3) Группировка данных

числов соответствии группа сточкахозяинхотеть С участием трехиндивидуальныйписьмочисло：groupByKey(),groupBy（）,cogroup（）。

• groupByKey(): этовстречаделатьиспользовать RDD серединаизключ Приходитьверночислов соответствии сруководитьточка Группа。верно Водининдивидуальный Зависит от типа K отключить типы V изценить Группастановитьсяиз RDD, результат из Узелфрукты RDD типвстречада[K, Iterable[V]] 。
• groupBy(): это Можеткиспользовать Веще нетстановитьсяверноизчислов соответствии Кон,также Можетккореньв соответствии с Кроме ключа те же условия для группы точек. это Можетк Получить одиночное число Письмо, к источнику RDD серединаиз Каждыйиндивидуальный Юаньбелыйделатьиспользовать Долженписьмочисло,Вернусь к Узелфруктыделатьдляключ снова группа точек.
• cogroup(): удалять Понятноверноодининдивидуальный RDD изчислов соответствии сруководитьточка Группа,возвращаться Можеткделатьиспользоватьодининдивидуальный Вызовделать cogroup() изписьмочисловерномногоиндивидуальныйобщийтакой жеодининдивидуальныйключиз RDD Провести точечную группу. Оба типа индивидуального включения предназначены для K иценитьизтипточка Недля V и W из RDD руководитьcogroup() час,придетсяприезжатьиз Узелфрукты RDD Типдля [(K, (Iterable[V], Iterable[W]))] 。нравитьсяфрукты Чтосерединаизодининдивидуальный RDD Для еще одного индивидуального RDD серединажитьсуществоватьизопределенныйиндивидуальныйключбезиметьверноотвечатьиз Записывать,Что Что?верноотвечатьизитератордлянулевой。cogroup() поставлять Понятнодлямногоиндивидуальный RDD продолжай считать соответствии группа сточкаизнаправление Закон。

Уведомление：

1. нравитьсяфруктытыобнаружил, что пишу Понятно Первыйделатьиспользовать groupByKey() Ранназад Сноваверноценитьделатьиспользовать reduce() или ВОЗ fold() изкод,тыоченьиметь Можетспособный Можеткпроходитьделатьиспользоватьодиндобрыйкореньв соответствии сключвыполнить агрегациюизписьмочисло Приходить Дажевысокийэффектземлявыполнитьтакой же Образецизэффектфрукты。ты Нетотвечать Долженделатьиспользоватьвперед ВОЗ,Потому что первое предполагает Понятное действие Держатьделать,Потребляет больше памяти,назад ВОЗТолькодаизменятьизменять Держатьделать。
2. cogroup() Неттолько Можеткиспользовать ВвыполнитьсоединятьДержатьделать,возвращаться Можеткиспользовать ПриходитьНайдите пересечение ключиз。удалятьэтот Изснаружи, cogroup() возвращатьсяспособныйтакой жечасотвечатьиспользовать Втрииндивидуальныйикна RDD。

(4)Подключение

Данные соединения могут быть pair RDD чаще всегоиспользоватьиз Держатьделать Изодин。соединятьнаправление Режим Различныймного Образец:верноснаружисоединять、левыйснаружисоединять、крестсоединятьк ИВнутрисоединять。

(5)Сортировка данных

Нам часто приходится RDD Поэтому расположены в обратном порядке. sortByKey() письмочислоперениматьодининдивидуальный Вызовделать ascending число параметра, указывающее, хотим ли мы хотеть сортировать Узелфрукты в порядке возрастания (по умолчанию ценитьдля true )。иметьчаснастакже Можетспособный Хочу закончить нажатие Полный Неттакой жеизсортироватьв соответствии св соответствии провести сортировку. хотеть держать В этом случае мы можемкпо дает определение буквенного числа сравнения.

существует Python серединак ​​строковая последовательность для целочисленного числа самоопределение сортировать:

rdd.sortByKey(ascending=True, numPartitions=None, keyfunc = lambda x: str(x))

существует Scala серединак ​​строковая последовательность для автоматизации целых чисел определение сортировать:

// скрытый Режимсортировать
неявное значение sortIntegersByString = новый Ordering[Int] {
    переопределить def Compare(a: Int, b: Int) = a.toString.compare(b.toString)
}
rdd.sortByKey()

4. Операция парного действия RDD

иизменятьизменять Держатьделать Такой же, основа всего есть RDD ветвьдержатьизпроходитьсистемадействие Держатьделатьтакже Всесуществовать pair RDD Доступно на. Пара RDDпоставлять Понятноодиннекоторый Лобснаружииздействие Держатьделать,Можеткпозволятьнасзаряжатьточкавыгодаиспользоватьчислов соответствии сизпара ключ-значениехарактеристика。этотнекоторый Держатьделать Списоксуществовать Понятнотаблица ниже：

5. Раздел данных

существоватьточкаткань Режимпрограммасередина,Стоимость связи очень высока,потому чтоэтотконтрольчислов соответствии сточкатканькполучать Получите минимумиз Сетевая передача Можетк极большойземля提升всетелесностьспособный。Толькоиметькогдачислов соответствии снабормного Второсортныйсуществоватьразличныйнравитьсясоединятьэтотдобрыйбаза Включиз Держатьделатьиспользуется вчас,точкаокругталантвстречаиметьпомощь。

Sparkизточкаокругнаправление Закон： Spark середина Местоиметьизпара ключ-значение RDD Все Можеткруководитьточкаокруг。системасистемавстречакореньв соответствии содининдивидуальныйДля ключаиз Книгачисловерно Юаньбелыйруководитьточкаокруг。хотя Spark безиметьдать управление дисплеем Каждыйиндивидуальныйключ Конкретносуществоватьгдеодининдивидуальныйработаделать Фестивальточкананаправление Закон(отделениеточка Оригиналпотому чтодаSpark т.е. делатьсуществовать некоторые узлы вышли из строя час еще Можетк делать), но Spark Можеткправильный Сохранятьтакой жеодинточкаокругизключ Появлятьсясуществоватьтакой жеодининдивидуальный Фестивальточканачальство。Сравниватьнравиться,ты Можетспособныйделатьиспользоватьха Редкийточкаокруг Воляодининдивидуальный RDD разделен на 100 индивидуальныйточкаокруг,этотчасключизха Редкийценитьверно100 Возьмите модуль Узелфруктытакой жеиз Записыватьвстречабыть освобожденнымсуществоватьодининдивидуальный Фестивальточканачальство。тытакже Можеткделатьиспользоватьобъемточкаокруг Закон,Воляключсуществоватьтакой жеодининдивидуальныйобъемокругмежду Внутрииз Записывать Всепомещатьсуществоватьтакой жеодининдивидуальный Фестивальточканачальство。

точкаокруг, RDD, узел родства：

• a single RDD has one or more partitions scattered across multiple nodes.
• a single partition is processed on a single node.
• a single node can handle multiple partitions.

Example: насточкаанализироватьтакодининдивидуальныйотвечатьиспользовать,этосуществовать Внутрижитьсередина Сохранятьжить Содиноткрытьоченьбольшойизиспользоватьсемьяинформацияповерхность——также Сразудаодининдивидуальный Зависит от (UserID, UserInfo) парная композиция из РДД, среди которых UserInfo Содержит список тем, на которые подписаны пользователи. Должен использоватьвстреча периодически объединять эту таблицу и отдельные небольшие файлы, для этого человека. альный небольшой файл середина хранит события, произошедшие за последние пять минут — по сути, Сразудаодининдивидуальный Зависит от (UserID, LinkInfo) парная композиция изповерхность,Сохраняет статус доступа каждого пользователя определенного веб-сайта за последние пять минут. Примернравиться,нас Можетспособныйнуждатьсяхотетьверноиспользоватьсемьядоступ Чтоеще нетподпискахозяинвопросизстраницаиз Состояниеруководитьсистемасчитать。нас Можеткделатьиспользовать Spark из join() Держатьделать Приходитьвыполнитьэтотиндивидуальный Группаобъединить Держатьделать,Чтосерединануждатьсяхотеть ПучокUserInfo и LinkInfo изиметьпоследовательностьвернокореньв соответствии с UserID Провести точечную группу. Наши изделия следует использоватьнравиться, как показано ниже:

// исходныйизменятькод;отHDFSнаодининдивидуальныйHadoop SequenceFileсередина Чтениеиспользовать информацию пользователя
// userDataсерединаиз Юаньбелыйвстречакореньв соответствии сэто считываются из источника «часиз Приходить», где распределяются блоки HDFS.
// В Spark в этот час нет Законаполучать Знать определенного индивидуального конкретного изUserID, соответствующего записи, на которой находится индивидуальный узел
val sc = new SparkContext(...)
val userData = sc.sequenceFile[UserID, UserInfo]("hdfs://...").persist()
// Периодические настройкииспользовать функцию число Приходвпричина из журнала событий, созданного за последние пять минут.
// Предположим, что этот даодининдивидуальный содержит (UserID, LinkInfo) в изSequenceFile
def processNewLogs(logFileName: String) {
    val events = sc.sequenceFile[UserID, LinkInfo](logFileName)
    val joined = userData.join(events)// RDD of (UserID, (UserInfo, LinkInfo)) pairs
    val offTopicVisits = joined.filter {
        case (userId, (userInfo, linkInfo)) => // Expand the tuple into its components
        !userInfo.topics.contains(linkInfo.topic)
    }.count()
    println("Number of visits to non-subscribed topics: " + offTopicVisits)
}

Этот код может работать корректно, но да Нетдостаточно эффективно. Вот почему processNewLogs() час Всевстречаиспользоватьприезжать join() Держатьделать,И нас не интересуют цифры по кругу По умолчанию.,соединять Держатьделатьвстреча Волядваиндивидуальныйчислов соответствии сSET серединаиз Все имеющиеключизха Редкийценить оказались Приходом, а Долженха Редкийценить то же самое по записи Он передается на ту же машину через сеть, а затем на этой машине обрабатываются все те же записи (см. рисунок). 4-4)。

потому чтодля userData В таблице журнала доступа появляется таблица, в которой отображаются часы с пятью точками. events хотетьбольшойпридетсямного,Местокхотетьнапрасно тратитьчасмежду做оченьмного Лобснаружиработаделать:существовать Каждый Второсортныйнастраиватьиспользоватьчас Всеверно userData Таблица ведется по узлам числов соответствии сshuffle, хотя эти числав соответствии сот Приходить Все Нетвстреча Изменятьизменять。

хотетьразвязать Решение этой одной проблемы тоже очень простое: существование программы запускаетсячас,верноuserData Использование таблицы partitionBy() Превратите этот стол в дляха Редкийточкаокруг. Автор: partitionBy Перейти на индивидуальный spark.HashPartitioner объект для реализации этой операции.

Пользовательский метод разделения Scala:

val sc = new SparkContext(...)
val userData = sc.sequenceFile[UserID, UserInfo]("hdfs://...")
    .partitionBy(new HashPartitioner(100)) // Строительство 100 индивидуальных точек округа
    .persist()

processNewLogs() направление Закон Может к Сохранять держать Нет Изменять,Зависит от Всуществовать Строить userData час настраивать использовать Понятно partitionBy() ,Spark Сразу Знать дорога Понятно Должен RDD да корень в соответствии с ключ из ха Редкий ценить Приходить точкаокругиз,таксуществоватьнастраиватьиспользовать join()час,Spark Сразувстречавыгодаиспользоватьприезжатьэтотодинточка。специфический Приходитьобъяснять,когданастраиватьиспользовать userData.join(events) час,Spark Тольковстречаверно events продолжай считать соответствии сперетасовать Держатьделать,Воля events средний специфический UserID из записи отправлено userData из соответствует точкаокругу на этой машине, нравиться, как показано ниже:

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

Q:для Разнообразный Что?точкаокруг ИзназадuserDataСразу Нетвстречапроисходитьперетасовать（shuffle）Понятно？ A:Первыйсмотретьодин Внизперетасоватьизопределение：Перемешать Spark для переиздания чиселв соответствии сиз-механизм, который облегчает этосуществование индивидуальнойточкиокружсерединаточка в нет той же группы. Обычно это вызвано наличием исполнительного механизма на машине. соответствии с,делатьпридетсяперетасоватьдаодининдивидуальныйсложныйиоткрытьприколотьоченьбольшойиз Держатьделать。насделатьиспользовать Понятноха Редкийточкаокругнаправление Режим,этовстреча Воля Инструментиметьтакой жеизkeyиз Юаньбелыйпомещатьприезжатьтакой жеодининдивидуальныйточкаокруг/точка Группа,также Сразудаобъяснять Нетжитьсуществовать Понятнодваиндивидуальныйточкаокругиметьтакой жеkeyиз Юаньбелыйиз Состояние,Местокjoinчас Сразу Нетвстреча Снова Второсортныйпроисходитьточка Группа,Нетвстречаиметьshuffleиз Держатьделать。（индивидуальныйлюдипричинаразвязать,Поправьте меня, если я ошибаюсь)

Уведомление： существовать Python середина,ты Нетспособный Воля HashPartitioner объект передан в partitionBy ,инуждатьсяхотеть Пучокнуждатьсяхотетьизточкаокругчислопередача Проходитьидти(примернравиться rdd.partitionBy(100) )。

(1) Получите метод разделения RDD.

существовать Scala и Java середина,ты Можеткделатьиспользовать RDD из partitioner Свойства (Java используется в partitioner() направление Закон)Приходитьполучать RDD стиль источкаокрунаправление. овстреча возврат одининдивидуальный scala.Option объект, этода Scala серединаиспользовать Приходитьжитьпомещать Можетспособныйжитьсуществоватьизвернослониз Класс контейнера。ты Можеткверноэтотиндивидуальный Option вернослоннастраиватьиспользовать isDefined() Приходитьисследовать Чтосерединаданетиметьценить,настраиватьиспользовать get() Приходитьполучать Чтосерединаизценить。нравитьсяфруктыжитьсуществоватьценитьизразговаривать,этотиндивидуальныйценитьвстречадаодининдивидуальный spark.Partitionerвернослон。этотприроданачальстводаодининдивидуальный Рассказыватьнас RDD серединакаждыйиндивидуальныйключточка Нерод Вгдеиндивидуальныйточкаокругизписьмочисло。

получать RDD източкаокругнаправление Режим:

scala> val pairs = sc.parallelize(List((1, 1), (2, 2), (3, 3)))
pairs: spark.RDD[(Int, Int)] = ParallelCollectionRDD[0] at parallelize at <console>:12

scala> pairs.partitioner
res0: Option[spark.Partitioner] = None

scala> val partitioned = pairs.partitionBy(new spark.HashPartitioner(2))
partitioned: spark.RDD[(Int, Int)] = ShuffledRDD[1] at partitionBy at <console>:14

scala> partitioned.partitioner
res1: Option[spark.Partitioner] = Some(spark.HashPartitioner@5147788d)

исходныйчасбезиметьточкаокругнаправление Режиминформация(одининдивидуальныйценитьдля None из Option объект). Тогда переходите к индивидуальному RDD руководитьха Редкийточкаокруг,создавать Понятно Нет.дваиндивидуальный RDD。

(2) Операции, которые выигрывают от разделения

Spark из Можетмного Держатьделать Всепредставлять Понятно Волячислов соответствии скореньв соответствии переключатель Процесс перемешивания выполняется между узлами. Все эти Держатьделать все встречиот число в соответствии с точка округ середина получать выгодный。Сразу Spark 1.0 и Слово, способный достаточно от число в соответствии с точка округ середина получать выгодный из Держать делать иметь cogroup() 、groupWith() 、 join() 、 leftOuterJoin() 、rightOuterJoin() 、 groupByKey() 、 reduceByKey() 、combineByKey() к И lookup() 。

1. Для чего-то вроде reduceByKey() так Толькоделатьиспользовать Водининдивидуальный RDD из Держатьделать,бегатьсуществоватьеще нетточкаокругиз RDD начасждатьвстречапривести к Каждыйиндивидуальныйключиз Местоиметьверноотвечатьценить Всесуществовать Каждый Рабочий столустройствоначальстворуководить本землясчитать Рассчитать,Просто хотеть передать локальную конечную редукцию каждого рабочего узла обратно на главный узел.,Таким образом, сетевые издержки изначально велики.
2. иверно Вразличныйнравиться cogroup() иjoin() soдва Юань'Hold'do',предварительно Первыйпродолжай считать соответствии сточкаокругвстречапривести к Чтосередина По меньшей мереодининдивидуальный RDD(делатьиспользоватьуже Знатьточкаокругустройствоиз Чтоиндивидуальный RDD)Нетпроисходитьчислов соответствии Перемешать.
3. нравитьсяфруктыдваиндивидуальный RDD делатьиспользоватьтакой же Образецизточкаокругнаправление Режим,ииэтоихвозвращатьсямедленныйжитьсуществоватьтакой же Образецизмашинаустройствоначальство(Сравниватьнравитьсяодининдивидуальный RDD дапасс mapValues() от Другойодининдивидуальный RDD серединасоздавать Приходитьиз,этотдваиндивидуальныйRDD Сразувстречавладетьиметьтакой жеизключиточкаокругнаправление Режим),или ВОЗЧтосерединаодининдивидуальный RDD Рассчитать не было выпущено, поэтому перекрестно-узловые вычисления соответствии сперетасовать Сразу Нетвстречапроисходить Понятно。

(3) Операции, влияющие на методы разделения

Spark Внутриотделение Знатьдорогакаждый Держатьделатьвстречанравитьсячто Влияниеточкаокругнаправление Режим,и Волявстречаверночислов соответствии сруководитьточкаокругиз Держатьделатьиз Узелфрукты RDD Автоматически устанавливать соответствующий источникокруглого устройства.。

изменятьизменять Держатьделатьиз Узелфруктыи Нетодин Конечновстречав соответствии суже Знатьизточкаокругнаправление Режимточкаокруг,этотчасвыходиз RDD Можетспособный Сразувстречабезиметьнастраиватьточкаокругустройство。примернравиться,когдатыверноодининдивидуальныйха Редкийточкаокругизпара ключ-значение RDD настраиватьиспользовать map() час,Зависит от перейти к map()изписьмочислопричина Аргументначальство Можеткизменять Изменять Юаньбелыйизключ,потому чтоэтот Узелфрукты Сразу Нетвстречаиметьзафиксированный Източкакруглое управление Режим。Spark Нетвстречаточкаанализироватьтыизписьмочисло Приходитьсуждениеключданетвстречаодеяло Сохранять Оставлять Приходить,потому чтои Даже Нетвстреча Сохранять Держать Извпереднастраивать Източкакруглое управление Режим。Нет Проходить,Spark поставлять Понятно Другойснаружидваиндивидуальный Держатьделать mapValues() иflatMapValues() делатьдлязаменятьнаправление Закон,этоих Можетк Сохранятьсертификат Каждыйиндивидуальный2-кортежизключ Сохранятьдержать Нет Изменять。

Местоиметьвстречадлягенерироватьиз Узелфрукты RDD Настраиватьточкаокругнаправление Режимиз Держатьделать: cogroup() 、 groupWith() 、join() 、 lef tOuterJoin() 、 rightOuterJoin() 、 groupByKey() 、 reduceByKey() 、combineByKey() 、 partitionBy() 、 sort() 、 mapValues() (нравитьсяфруктыотец RDD иметьточкаокругнаправление Режимизразговаривать)、flatMapValues() (нравитьсяфруктыотец RDD иметьточкаокругнаправление Режимизразговаривать),к И filter() (нравитьсяфруктыотец RDD иметьточкаокругнаправление Режимизразговаривать)。Чтоон Местоиметьиз Держатьделатьгенерироватьиз Узелфрукты Все Нетвстречажитьсуществоватьособенный Конечно Източкакруглое управление Режим。

верно Вдва Юань'Держат'делать',выходчислов соответствии сизточкаокружнаправление зависит от родителя RDD Източкакруглое управление Режим。по умолчанию Состояние Вниз,Узелфруктывстреча Выбиратьиспользоватьха Редкийточкаокруг,точкаокругизчислоколичествои Держатьделатьизпараллельныйстепеньодин Образец。Нет Проходить,нравитьсяфрукты Чтосерединаизодининдивидуальныйотец RDD уженастраивать Проходитьточкаокругнаправление Режим,Что Что?Узелфрукты Сразувстреча Выбиратьиспользовать Чтодобрыйточкаокругнаправление Режим;нравитьсяфруктыдваиндивидуальныйотец RDD Все в стиле точкаокругнаправление,Узелфрукты. RDD встреча Выбиратьиспользовать Нет.одининдивидуальныйотец RDD Източкакруглое управление Режим。

(4)Пример: PageRank

PageRankРассчитать Закондак Google из Ларри· Пегги (Ларри) Page)изименованиеиз,использовать Приходитькореньв соответствии сснаружиотделениедокументориентированныйодининдивидуальныйдокументиз Связь,вернонаборобъединитьсередина Каждыйиндивидуальныйдокументиз Тяжелыйхотетьстепень пожертвованияодининдивидуальныйстепеньколичествоценить。Должен Рассчитать Закон Можеткиспользовать ВСортировка веб-страниц,конечно,также Можеткиспользовать ВсортироватьСтатьи о науке и технологияхилисоциальная сетьсерединаиметь Влияниеизиспользоватьсемья。

PageRank даосуществлятьмного Второсортныйсоединятьизодининдивидуальный Итерировать Рассчитать Закон,потому чтоэтотэтода RDD точкаокруг Держатьделатьизодининдивидуальныйоченьхорошийизиспользоватьпример。Рассчитать Законвстречаподдерживатьдваиндивидуальныйчислов соответствии снабор:одининдивидуальный Зависит от(pageID, linkList) из Юаньбелый состоит из Каждыйиндивидуальный страницы из соседних страниц из списка другойиндивидуальный Зависит; от (pageID, rank) Юаньбелый Группастановиться,Включать Каждыйиндивидуальныйстраницаизкогдавпередсортироватьверноценить。этов соответствии снравиться Внизшагруководитьсчитать Рассчитать：

• (1) Воля Каждыйиндивидуальныйстраницаизсортироватьценитьисходныйизменятьдля 1.0。
• (2) существуют Каждый итерации середина, на страницу p ,КЧто Каждыйиндивидуальный Взаимно邻страница(иметьпрямой Связьизстраница)отправлятьодининдивидуальныйценитьдляrank(p)/numNeighbors(p) изспособствоватьценить。
• (3) Воля Каждыйиндивидуальныйстраницаизсортироватьценитьнастраиватьдля 0.15 + 0.85 * contributionsReceived 。

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

Scala версия PageRank:

// Предполагая список соседних страниц кSpark objectFileиз хранилища форм
val links = sc.objectFile[(String,Seq[String])]("links")
                .partitionBy(new HashPartitioner(100))
                .persist()
// Воля Каждыйиндивидуальныйстраницаизсортироватьценитьисходныйизменятьдля1.0;Зависит от ВделатьиспользоватьmapValues,генерироватьизRDD
// източкаокругнаправление Режимвстречаи"links"изодин Образец
var ranks = links.mapValues(v => 1.0)
// Проведите 10 итераций PageRank
for(i <- 0 until 10) {
    val contributions = links.join(ranks).flatMap {
        case (pageId, (links, rank)) =>
            links.map(dest => (dest, rank / links.size))
    }
    ranks = contributions.reduceByKey((x, y) => x + y).mapValues(v => 0.15 + 0.85*v)
}
// Напишите окончательный рейтинг
ranks.saveAsTextFile("ranks")

Уведомление：

• Когда мы создаём впервые ranks час,насделатьиспользовать mapValues() и Нетда map() Приходить Сохранять Держатьотец RDD( links )източкаокругнаправление Режим,такверноэторуководитьиз Нет.один Второсортныйсоединять Держатьделать Сразувстречаоткрытьприколотьочень Маленький。
• существовать Тело петлисередина,нассуществовать reduceByKey() назадделатьиспользовать mapValues() ;потому чтодля reduceByKey() из Узелфруктыужедаха Редкийточкаокругиз Понятно,такодин Приходить,Внизодин Второсортныйциклсередина Волякартографирование Держатьделатьиз Узелфрукты Снова Второсортныйи links руководитьсоединять Держатьделатьчас Сразувстреча Даже加высокийэффект。

для Понятномаксимумизменятьточкаокруг Связанныйотличныйизменятьизскрытыйсуществоватьделатьиспользовать,тыотвечать Долженсуществоватьниктонуждатьсяизменять Изменять Юаньбелыйизключчасвыхлопколичестводелатьиспользовать mapValues() или flatMapValues() 。

(5) Индивидуальный метод разделения

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

В качестве примера индивидуального предположим, что мы запускаем раздел «бывший один» серединаиз на веб-странице «хотетьсуществоватьиндивидуальный из». PageRank Рассчитать Закон。существоватьэтотвнутри,Каждыйиндивидуальныйстраницаиз ID(RDD серединаизключ)дастраницаиз URL。когданасделатьиспользовать Простойизха Редкийписьмочислоруководитьточкаокругчас,владетьиметьсходствоиз URL изстраница(Сравниватьнравиться http://www.cnn.com/WORLD и http://www.cnn.com/US)Можетспособныйвстречаодеялоточкаприезжатьнад Полный Неттакой жеиз Фестивальточканачальство。Рани,нас Знатьдорогасуществоватьтакой жеодининдивидуальныйдоменное имя Внизизвеб-страница Дажеиметь Можетспособный Взаимно互Связь。Зависит от В PageRank Необходимо иметькаждый итерации серединаот Каждыйиндивидуальный страницу, чтобы она имела соседние и З страницы отправляют одно сообщение, поэтому лучше объединить эти страницы в одну группу. Можеткделатьиспользовать с момента определенияизточкаокруга Приходить реализация только кореньв соответствии доменное имя Нетдавсеиндивидуальный URL Приходитьточкаокруг。

Scalaсередина： хотетьвыполнитьсопределениеизточкаокругустройство,тынуждатьсяхотетьнаследовать org.apache.spark.Partitionerдобрыйивыполнитьподтрииндивидуальныйнаправление Закон：

• numPartitions: Int :возвращатьсясоздавать Приходитьизточкаокругчисло。
• getPartition(key: Any): Int :возвращатьсяданныйключизточкаокругсерийный номер(0 приезжатьnumPartitions-1 )。
• equals() :Java Суждение равенства из критериев направления Закон. Эта реализация очень тяжелая, хотя Spark Нужно хотетьиспользовать это индивидуальное направление Закон П риходить проверяет, совпадает ли объект-контейнер тыизточкаокруг с другими экземплярами контейнера точкаокруг, например: Spark талант МожетксуждениедваиндивидуальныйRDD източкаокругнаправление Режимданеттакой же。

делатьиспользоватьсопределениеиз Partitioner это просто: просто передай хотеть этому partitionBy() направление Закон Прямо сейчас Может。

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

Scala сопределениеточкаокругнаправление Режим:

class DomainNamePartitioner(numParts: Int) extends Partitioner { 
   
    override def numPartitions: Int = numParts
    override def getPartition(key: Any): Int = {
        val domain = new Java.net.URL(key.toString).getHost()
        val code = (domain.hashCode % numPartitions)
        if(code < 0) {
            code + numPartitions // делать его неотрицательным
        }else{
            code
        }
    }
    // использовать ПриходитьпозволятьSparkокругточкаточкаокругписьмочисловернослонизJava equalsнаправление Закон
    override def equals(other: Any): Boolean = other match {
        case dnp: DomainNamePartitioner => dnp.numPartitions == numPartitions
        case _ => false
    }
}

Уведомление： 1. Когда тыиз Рассчитать Закон зависит от Java из hashCode() направление Закончас,этотиндивидуальныйнаправление Закониметь Можетспособныйвстречавозвращатьсягрузчисло。тынуждатьсяхотетьдесятьточкаосторожный,правильный Сохранять getPartition() Всегда возвращает единичное неотрицательное число. 2. нассуществовать equals() направление Законсередина,делатьиспользовать Scala из шаблона соответствует символу Держатьделать ( match )Приходитьисследовать other данетдаDomainNamePartitioner , и существование установлено, когда автоматически выполняется преобразование типов.

Pythonсередина：

существовать Python середина,Нетнуждатьсяхотеть Расширять Partitioner добрый,ида Пучокодининдивидуальныйособенный Конечноизха Редкийписьмочислоделатьдляодининдивидуальный Лобснаружиизженьшеньчислопроходить给 RDD.partitionBy() письмочисло。

Python сопределениеточкаокругнаправление Режим:

import urlparse

def hash_domain(url):
    return hash(urlparse.urlparse(url).netloc)

rdd.partitionBy(20, hash_domain) # Создать 20индивидуальныйточкаокружок

Уведомление: этотвнутриты Местопроходить Проходитьидтиизха Редкийписьмочисловстречаодеялои Чтоон RDD източкаокругписьмочислоокругточкаоткрыть Приходить。нравитьсяфруктытыдуматьхотетьверномногоиндивидуальный RDD делатьиспользоватьтакой жеизточкаокругнаправление Режим,Сразуотвечать Долженделатьиспользоватьтакой жеодининдивидуальныйписьмочисловернослон,Сравниватьнравитьсяодининдивидуальный Полныйбюрописьмочисло,и Нетдадля Каждыйиндивидуальный RDD создаватьодининдивидуальныйновыйизписьмочисловернослон。

Ref: «Быстрая искра большого числа» соответствии Анализ стока》

Автор: Полный стек программист стек лидер,Пожалуйста, укажите источник для перепечатки: https://javaforall.cn/126622.html Оригинальная ссылка: https://javaforall.cn