Рдд 03 схема подключения: РДД-03. Изготовитель оставляет за собой право вносить изменения в конструкцию и принципиальную схему изделия, не ухудшающие его характеристик.

1. Сначала Spark анализирует запрос и создает неразрешенный логический план
   • Проверяет синтаксис запроса.
   • Не проверяет семантику, означающую наличие имени столбца, типов данных.
2. Analysis: Используя Catalyst, он преобразует неразрешенный логический план в разрешенный логический план , также известный как логический план.
   1. Каталог содержит имена столбцов и типы данных, на этом этапе он проверяет столбцы, упомянутые в запросе с каталогом.
3. Оптимизация: преобразует логический план в оптимизированный логический план .
4. Планировщик: Теперь он создает Один или несколько физических планов из оптимизированного логического плана.
5. Модель затрат: На этом этапе рассчитывается стоимость для каждого физического плана и выбирается Лучший физический план .
6. Создание RDD: генерируется RDD, это заключительный этап оптимизации запросов, который генерирует RDD в байт-коде Java.

После генерации RDD в байтовом коде механизм выполнения Spark выполняет преобразования и действия.

Что такое адаптивное выполнение запросов

Adaptive Query Optimization в Spark 3.0, повторно оптимизирует и корректирует планы запросов на основе метрик времени выполнения, собранных во время выполнения запроса, эта повторная оптимизация плана выполнения происходит после каждого этапа запроса, поскольку этап дает правильное место для повторного выполнения. оптимизация.

Примечание: В задании Spark этап создается с каждым более широким преобразованием, в котором происходит перетасовка данных.

Как это развивалось?

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

• Spark 1.x - Добавлен оптимизатор Catalyst и вольфрамовый механизм выполнения
• Spark 2.x - Добавлен оптимизатор на основе затрат
• Spark 3.0 - теперь добавлено адаптивное выполнение запросов

Включение адаптивного выполнения запросов

Адаптивное выполнение запросов по умолчанию отключено. Чтобы включить, установите для свойства конфигурации spark.sql.adaptive.enabled значение true . Помимо этого свойства, вам также необходимо включить функцию AQE, которую вы собираетесь использовать, которая будет объяснена позже в этом разделе.

 
spark.conf.set ("spark.sql.adaptive.enabled", истина)
  

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

Spark SQL UI

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

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

Функции выполнения адаптивных запросов

Spark 3.0 включает в себя три основных функции AQE.

• Уменьшение разделов после перемешивания.
• Коммутационные стратегии объединения для широковещательного присоединения
• Оптимизация перекосного соединения

Уменьшение количества разделов после перемешивания.

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

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

 
spark.conf.set ("spark.sql.adaptive.enabled", истина)
spark.conf.set ("spark.sql.adaptive.coalescePartitions.enabled", истина)
  

Теперь давайте посмотрим, как это работает, сначала без включения AQE.

 
  импортировать spark.implicits._
  val simpleData = Seq (("Джеймс", "Продажи", "Нью-Йорк", 
,34,10000),
    («Майкл», «Сейлз», «Нью-Йорк», 86000,56,20000),
    («Роберт», «Продажи», «Калифорния», 81000,30,23000),
    («Мария», «Финанс», «CA», 
,24,23000),
    («Раман», «Финансы», «CA», 99000,40,24000),
    («Скотт», «Финансы», «Нью-Йорк», 83000,36,19000),
    («Джен», «Финансы», «Нью-Йорк», 79000,53,15000),
    («Джефф», «Маркетинг», «Калифорния», 80000,25,18000),
    («Кумар», «Маркетинг», «Нью-Йорк», ,50,21000)
  )
  val df = simpleData.toDF ("имя сотрудника", "отдел", "штат", "зарплата", "возраст", "премия")
  
  val df1 = df.groupBy ("отдел"). count ()
  println (df1.rdd.getNumPartitions)
  

Поскольку groupBy () запускает более широкое преобразование или перемешивание, оператор df1.rdd.getNumPartitions приводит к созданию 200 разделов; Это связано с тем, что Spark по умолчанию создает 200 разделов для операций перемешивания.

Теперь давайте запустим тот же пример после включения AQE

 
  Искра.conf.set ("spark.sql.adaptive.enabled", истина)
  val df2 = df.groupBy ("отдел"). count ()
  println (df2.rdd.getNumPartitions)
  

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

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

Переключение стратегий присоединения для широковещательного присоединения

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

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

Оптимизация стыковки с перекосом

Иногда мы можем встретить данные в разделах, которые распределены неравномерно, это называется перекосом данных. Такие операции, как соединение, выполняются на этих разделах очень медленно.Включив AQE, Spark проверяет статистику этапа и определяет, есть ли какие-либо соединения с перекосом, и оптимизирует его, разделяя большие разделы на меньшие (соответствующий размеру раздела в другой таблице / фрейме данных).

 
spark.conf.set ("spark.sql.adaptive.enabled", истина)
spark.conf.set ("spark.sql.adaptive.skewJoin.enabled", истина)
  

Убедитесь, что вы установили для обоих свойств конфигурации значение true.

Номер ссылки

Понимание RDD Spark - Часть 3 | пользователя Anveshrithaa S | Аналитика Vidhya

Добро пожаловать! В предыдущем блоге был представлен краткий обзор RDD в Spark.В этой статье мы обсудим две важные операции с RDD Spark - преобразования и действия вместе с примерами. К концу этого блога у вас будет четкое представление о RDD и о том, как писать программы PySpark с использованием RDD.

Создание RDD в PySpark

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

Существует несколько способов создания RDD. Один из простых способов - распараллелить существующую коллекцию в программе драйвера, передав ее методу SparkContext parallelize () . Здесь элементы коллекции копируются в RDD и могут обрабатываться параллельно.

 data = [«Scala», «Python», «Java», «R»] myRDD = sc.parallelize (data) 

Примечание: Не забудьте создать SparkContext sc (если вы не используя оболочку PySpark, которая автоматически создает sc).Это первое, что вы должны сделать при написании любой программы на Spark.

Здесь данные в параллельной коллекции разделяются на разделы, и на каждом разделе кластера будет выполняться одна задача. По умолчанию Spark устанавливает количество разделов в зависимости от кластера. Его также можно установить вручную, передав количество разделов в качестве второго параметра методу parallelize () .

 data = [«Scala», «Python», «Java», «R»] # данные разделены на два раздела 
 myRDD = sc.parallelize (data, 2) 

Другой способ создания Spark RDD - из других источников данных, таких как локальная файловая система, Cassandra, HDFS и т. д. Здесь данные загружаются из внешних наборов данных. Для этого мы используем метод SparkContext textFile , который принимает URL-адрес файла в качестве параметра.

 # текстовый файл в RDD 
 myRDD = sc.textFile («/ path_to_file / textdata.txt») # CSV-файл в RDD 
 myRDD = sc.textFile («/ path_to_file / csvdata.csv») 

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

СДР Spark поддерживают два типа операций, а именно преобразования и действия. После создания RDD мы можем выполнять преобразования и действия с ними.

Преобразования - это операции над RDD, которые создают новый RDD путем внесения изменений в исходный RDD. Вкратце, это функции, которые принимают существующий RDD в качестве входных данных для предоставления новых RDD в качестве выходных данных без внесения изменений в исходный RDD (обратите внимание, что RDD неизменяемы!) Этот процесс преобразования RDD в новый выполняется с помощью таких операций, как фильтр, карта, reduceByKey, sortBy и т. д.Как видно из предыдущего блога, RDD следуют ленивой оценке. То есть преобразования RDD не будут выполняться до тех пор, пока они не будут запущены при необходимости. Таким образом, эти операции можно выполнить в любое время, просто вызвав действие над данными. Преобразования в RDD можно разделить на две категории: узкие и широкие.

• В узких преобразованиях результат преобразования таков, что в выходном RDD каждый из разделов имеет записи из одного и того же раздела в родительском RDD.Такие операции, как Map, FlatMap, Filter, Sample, подвергаются узким преобразованиям.
• Тогда как в широких преобразованиях данные в каждом из разделов результирующего RDD поступают из нескольких разных разделов в родительском RDD. Функции преобразования, такие как groupByKey (), reduceByKey (), подпадают под категорию широких преобразований.

Давайте посмотрим на некоторые преобразования в RDD.

map ()

map () возвращает новый СДР, применяя функцию к каждому из элементов в исходном СДР.

 data = [1, 2, 3, 4, 5] myRDD = sc.parallelize (data) # Возвращает новый RDD, умножая все элементы родительского RDD на 2 
 newRDD = myRDD.map (lambda x: x * 2 ) print (newRDD.collect ()) 

Вывод:

 [2, 4, 6, 8, 10] 

flatMap ()

flatMap () возвращает новый RDD, применяя функцию к каждому элементу родительского RDD, а затем сглаживает результат. Давайте посмотрим на примере, чтобы понять разницу между map () и flatMap () .

 data = [1, 2, 3] 
 myRDD = sc.parallelize (data)  #map () возвращает [[1], [1, 2], [1, 2, 3]]  
 mapRDD = myRDD .map (lambda x: range (1, x))  #flatmap () возвращает [1, 1, 2, 1, 2, 3]  
 flatMapRDD = myRDD.flatMap (lambda x: range (1, x)) 

filter ()

filter () возвращает новый RDD, содержащий только элементы в родительском RDD, которые удовлетворяют функции внутри filter.

 data = [1, 2, 3, 4, 5, 6] myRDD = sc.parallelize (data) # возвращает RDD только с элементами, которые делятся на 2 
 newRDD = myRDD.filter (lambda x: x% 2 == 0) print (newRDD.collect ()) 

Вывод:

 [2, 4, 6] 

отдельный ()

отдельный () возвращает новый RDD, содержащий только отдельные элементы в родительском RDD

 data = [1, 2, 2, 3, 3, 3] myRDD = sc.parallelize (data) newRDD = myRDD.distinct () print (newRDD.collect ()) 

Выход:

 [1, 2, 3] 

groupByKey ()

groupByKey () группирует значения для каждого ключа в парах (ключ, значение) RDD в одиночная последовательность.Мы получаем объект, который позволяет нам перебирать результаты.

 myRDD = sc.parallelize ([(«a», 1), («a», 2), («a», 3), («b», 1)]) # распечатать результат в виде списка 
 resultList = myRDD.groupByKey (). mapValues ​​(список) reultList.collect () 

Вывод:

 [('a', [1, 2, 3]), ('b', [1])] 

reduceByKey ()

reduceByKey () при вызове набора данных из пар (ключ, значение) возвращает новый набор данных, в котором агрегированы значения для каждого из его ключей.

 from operator import addmyRDD = sc.parallelize ([(«a», 1), («a», 2), («a», 3), («b», 1)]) # добавляет значения на keys 
 newRDD = myRDD.reduceByKey (add) newRDD.collect () 

Вывод:

 [('a', 6), ('b', 1)] 

Примечание: groupByKey () и reduceByKey () могут показаться похожими. Разница между ними в том, что при вызове groupByKey в RDD данные в разделах перетасовываются по сети и только после этого группируются.Это приводит к ненужной передаче большого количества данных по сети. Принимая во внимание, что reduceByKey () объединяет пары на основе ключа на их соответствующих разделах локально, прежде чем перетасовать данные, что делает его более эффективным. Таким образом, reduceByKey () используется для больших наборов данных для повышения производительности.

sortByKey ()

sortByKey () возвращает новый RDD с парами (ключ, значение) родительского RDD в отсортированном порядке в соответствии с ключом.

 myRDD = sc.parallelize ([(«c», 1), («d», 2), («a», 3), («b», 4)]) # сортировать по ключу 
 newRDD = myRDD .sortByKey () newRDD.collect () 

Вывод:

 [('a', 3), ('b', 4), ('c', 1), ('d', 2) ] 

union ()

union () возвращает новый RDD, который является объединением родительских RDD.

 myRDD1 = sc.parallelize ([1, 2, 3, 4]) myRDD2 = sc.parallelize ([3, 4, 5, 6, 7]) # объединение myRDD1 и myRDD2 
 newRDD = myRDD1.union (myRDD2 ) newRDD.collect () 

Вывод:

 [1, 2, 3, 4, 3, 4, 5, 6, 7] 

Точно так же у нас есть crossction (), который возвращает пересечение двух RDD.

Действия - это операции с RDD для выполнения вычислений и возврата окончательного результата обратно драйверу. Он запускает выполнение для выполнения промежуточных преобразований после загрузки данных в RDD и, наконец, передачи результата обратно. Эти действия представляют собой операции, которые приводят к значениям, не относящимся к RDD. Как мы уже видели, преобразования выполняются только тогда, когда действие требует возврата результата. Сбор, сокращение, countByKey, count - вот некоторые из действий.

Давайте рассмотрим некоторые действия с RDD на примерах.

collect ()

Глядя на приведенные выше примеры, вы теперь знаете, что делает collect () . Он возвращает список, содержащий все элементы СДР.

count ()

count () возвращает количество элементов в RDD

 data = [«Scala», «Python», «Java», «R»] myRDD = sc.parallelize (data) # Возвращает 4 в качестве вывода. 
 myRDD.count () 

reduce ()

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

 data = [1, 2, 3, 4, 5] myRDD = sc.parallelize (data) # возвращает произведение всех элементов 
 myRDD.reduce (lambda x, y: x * y) 

Вывод:

 120 

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

 data = [«Scala», «Python», «Java», «R»] myRDD = sc.parallelize (data) # Объединить строковые элементы 
 myRDD.reduce (lambda x, y: x + y) 

Вывод:

 'ScalaPythonJavaR' 

foreach ()

Применяет функцию к самому элементу в RDD

 def fun (x): 
 print (x) data = [«Scala», «Python», « Java »,« R »] myRDD = sc.parallelize (data) # функция, применяемая ко всем элементам 
 myRDD.foreach (fun) 

Вывод:

 Scala 
 Python 
 Java 
 R 

countByValue ()

Возвращает количество каждого уникального значения в RDD как словарь с парами (значение, количество).

 data = [«Python», «Scala», «Python», «R», «Python», «Java», «R»,] myRDD = sc.parallelize (data) #items () возвращает список с все ключи и значения словаря, возвращаемые функцией countByValue () myRDD.countByValue (). items () 

Вывод:

 [('Python', 3), ('R', 2), ('Java', 1), ('Scala', 1)] 

countByKey ()

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

 data = [(«a», 1), («b», 1), («c», 1), («a», 1)] myRDD = sc.parallelize (data) myRDD.countByKey () .items () 

Вывод:

 [('a', 2), ('b', 1), ('c', 1)] 

take (n)

take (n) возвращает первые n элементов RDD в том же порядке.

 data = [2, 5, 3, 8, 4] myRDD = sc.parallelize (data) #return the first 2 element 
 myRDD.take (3) 

Вывод:

 [2, 5, 3] 

top (n)

Возвращает первые n элементов из всех RDD элементы расположены в порядке убывания.

 data = [2, 5, 3, 8, 4] myRDD = sc.parallelize (data) # вернуть первые 2 элемента 
 myRDD.take (3) 

Вывод:

 [8, 5, 4 ] 

Подробнее о внутреннем устройстве и архитектуре Spark

, автор - Джайвардхан Редди

Apache Spark - это распределенная универсальная среда кластерных вычислений с открытым исходным кодом. Приложение Spark - это процесс JVM, в котором выполняется пользовательский код с использованием Spark в качестве сторонней библиотеки.

В рамках этого блога я покажу, как Spark работает с архитектурой Yarn, на примере и различных задействованных фоновых процессах, таких как:

• Контекст Spark
• Менеджер ресурсов пряжи, Мастер приложений и запуск исполнители (контейнеры).
• Настройка переменных среды, ресурсов задания.
• CoarseGrainedExecutor Backend и RPC на основе Netty.
• SparkListeners.
• Выполнение задания (Логический план, Физический план).
• Spark-WebUI.

Контекст Spark - это точка входа первого уровня и сердце любого приложения Spark. Spark-shell - это не что иное, как REPL на основе Scala с двоичными файлами искры, который создаст объект sc, называемый контекстом искры.

Мы можем запустить искровую оболочку, как показано ниже:

  spark-shell --master yarn \ --conf spark.ui.port = 12345 \ --num-executors 3 \ --executor-cores 2 \ - execor-memory 500M  

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

Или вы можете запустить Spark Shell, используя конфигурацию по умолчанию.

  Spark-Shell --master Yarn  

Конфигурации присутствуют как часть Spark-env.sh

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

Доступ к объекту контекста искры можно получить с помощью sc.

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

После создания контекста Spark он сверится с Cluster Manager и запустит Application Master , то есть запускает контейнер и регистрирует обработчики сигналов .

После запуска мастера приложений он устанавливает соединение с драйвером.

Затем ApplicationMasterEndPoint запускает прокси-приложение для подключения к диспетчеру ресурсов.

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

ii) YarnRMClient зарегистрируется в Мастере приложений.

iii) YarnAllocator: запросит 3 контейнера исполнителей, каждый с 2 ​​ядрами и 884 МБ памяти, включая 384 МБ служебных данных

iv) AM запускает поток репортера

Теперь средство распределения пряжи получает токены от драйвера для запуска узлов Executor и запуска контейнеры.

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

Среда выполнения Spark (SparkEnv) - это среда выполнения со службами Spark, которые используются для взаимодействия друг с другом с целью создания распределенной вычислительной платформы для приложения Spark.

Контекст запуска исполнителя YARN назначает каждому исполнителю идентификатор исполнителя для идентификации соответствующего исполнителя (через Spark WebUI) и запускает CoarseGrainedExecutorBackend.

После получения ресурсов из Resource Manager мы увидим запуск исполнителя

CoarseGrainedExecutorBackend - ExecutorBackend, который управляет жизненным циклом отдельного исполнителя. Он отправляет водителю статус исполнителя.

При запуске ExecutorRunnable CoarseGrainedExecutorBackend регистрирует конечную точку RPC Executor и обработчики сигналов для связи с драйвером (т.е.е. с конечной точкой RPC CoarseGrainedScheduler) и сообщить, что он готов к запуску задач.

Netty-based RPC - Он используется для связи между рабочими узлами, контекстом искры, исполнителями.

NettyRPCEndPoint используется для отслеживания статуса результата рабочего узла.

RpcEndpointAddress - это логический адрес конечной точки, зарегистрированной в среде RPC, с RpcAddress и именем.

Он имеет формат, показанный ниже:

Это первый момент, когда CoarseGrainedExecutorBackend инициирует обмен данными с драйвером, доступным по адресу driverUrl, через RpcEnv.

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

SparkContext запускает LiveListenerBus, который находится внутри драйвера. Он регистрирует JobProgressListener в LiveListenerBus, который собирает все данные для отображения статистики в пользовательском интерфейсе искры.

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

Spark поставляется с двумя слушателями, которые демонстрируют большинство действий

i) StatsReportListener

ii) EventLoggingListener

EventLoggingListener: часть сервера истории Spark, вы можете обрабатывать данные журнала событий.Журнал событий Spark записывает информацию об обработанных заданиях / этапах / задачах. Его можно включить, как показано ниже ...

Файл журнала событий можно прочитать, как показано ниже.

• Драйвер Spark регистрируется в показателях рабочей нагрузки / производительности задания в каталоге spark.evenLog.dir в виде файлов JSON.
• Существует один файл для каждого приложения, имена файлов содержат идентификатор приложения (следовательно, включая временную метку) application_1540458187951_38909.

Показывает тип событий и количество записей для каждого.

Теперь давайте добавим StatsReportListener к spark.extraListeners и проверим статус задания.

Включите уровень ведения журнала INFO для регистратора org.apache.spark.scheduler.StatsReportListener, чтобы просматривать события Spark.

Чтобы включить прослушиватель, вы регистрируете его в SparkContext. Это можно сделать двумя способами.

i) Использование метода SparkContext.addSparkListener (слушатель: SparkListener) внутри вашего приложения Spark.

Щелкните ссылку для реализации настраиваемых прослушивателей - CustomListener

ii) Использование параметра командной строки conf

Давайте прочитаем образец файла и выполним операцию подсчета, чтобы увидеть StatsReportListener.

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

i) P Распараллеливание существующей коллекции в вашей программе драйвера

ii) Ссылка на набор данных во внешней системе хранения

RDD создаются либо с использованием файла в файловой системе Hadoop, либо с помощью существующего Коллекция Scala в программе драйвера и ее преобразование.

Давайте возьмем образец фрагмента, как показано ниже.

Выполнение указанного выше фрагмента происходит в 2 этапа.

6.1 Логический план: На этом этапе RDD создается с использованием набора преобразований. Он отслеживает эти преобразования в программе драйвера, создавая вычислительную цепочку (серию RDD) в виде графика преобразований. для создания одного RDD под названием Lineage Graph .

Преобразования можно разделить на 2 типа

• Узкое преобразование: Конвейер операций, которые могут выполняться как один этап и не требуют перетасовки данных по разделам - например, карта, фильтр и т. Д. ..

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

• Широкое преобразование: Здесь каждая операция требует перетасовки данных, отныне для каждого широкого преобразования будет создаваться новый этап - например, reduceByKey и т. Д.

Мы можем просмотреть граф происхождения, используя toDebugString

6.2 Физический план: На этом этапе, как только мы запускаем действие в RDD, DAG Scheduler смотрит на происхождение RDD и предлагает лучший план выполнения с этапами и задачи вместе с TaskSchedulerImpl и параллельно выполнять задание в виде набора задач.

После выполнения операции действия SparkContext запускает задание и регистрирует RDD до первого этапа (то есть до любых широких преобразований) как часть DAGScheduler.

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

В случае отсутствия задач назначает задачи исполнителям.

Каждая задача назначается CoarseGrainedExecutorBackend исполнителя.

Получает информацию о блоке из Namenode.

, он выполняет вычисление и возвращает результат.

Затем DAGScheduler ищет новые запускаемые этапы и запускает операцию следующего этапа (reduceByKey).

ShuffleBlockFetcherIterator получает блоки для перемешивания.

Теперь операция сокращения разделена на 2 задачи и выполняется.

По завершении каждой задачи исполнитель возвращает результат драйверу.

После завершения задания отображается результат.

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

Мы увидим визуализацию Spark-UI как часть предыдущего шага 6 .

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

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

Вы можете увидеть время выполнения каждого этапа.

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

Далее, мы можем щелкнуть вкладку «Executors», чтобы просмотреть используемые Executor и драйвер.

Теперь, когда мы увидели, как Spark работает внутри, вы можете определить поток выполнения, используя Spark UI, журналы и настраивая Spark EventListeners для определения оптимального решения при отправке задания Spark.

Примечание : Команды, которые были выполнены в связи с этим сообщением, добавлены как часть моей учетной записи GIT.

Точно так же вы также можете прочитать больше здесь:

Если вы тоже хотите, вы можете связаться со мной в LinkedIn - Jayvardhan Reddy.

Если вам понравилось это читать, вы можете нажать на хлопок и сообщить об этом другим. Если вы хотите, чтобы я добавил что-нибудь еще, не стесняйтесь оставлять ответ?

История трех API-интерфейсов Apache Spark

Из всех радостей разработчиков нет ничего более привлекательного, чем набор API-интерфейсов, которые делают разработчиков продуктивными, простыми в использовании, интуитивно понятными и выразительными.Apache Spark привлекает разработчиков простыми в использовании API-интерфейсами для работы с большими наборами данных на разных языках: Scala, Java, Python и R.

В этом блоге я исследую три набора API - RDD, DataFrames и Datasets - доступных в Apache Spark 2.2 и последующих версиях; почему и когда следует использовать каждый набор; описать их эффективность и преимущества оптимизации; и перечислить сценарии использования DataFrames и Datasets вместо RDD. В основном я сосредоточусь на DataFrames и Datasets, потому что в Apache Spark 2.0 эти два API объединены.

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

Устойчивый распределенный набор данных (RDD)

RDD был основным пользовательским API в Spark с момента его создания. По сути, RDD - это неизменяемая распределенная коллекция элементов ваших данных, распределенная по узлам в вашем кластере, которые могут работать параллельно с низкоуровневым API, который предлагает преобразований и действий .

Когда использовать RDD?

Рассмотрите эти сценарии или распространенные варианты использования RDD, когда:

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

Что происходит с RDD в Apache Spark 2.0?

Вы можете спросить: Выносятся ли RDD в категорию граждан второго сорта? Они устарели?

Ответ: НЕТ!

Более того, как вы заметите ниже, вы можете легко перемещаться между DataFrame или Dataset и RDD по своему желанию - с помощью простых вызовов методов API, а DataFrames и Datasets построены на основе RDD.

DataFrames

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

В нашем предварительном просмотре вебинара Apache Spark 2.0 и в последующем блоге мы упоминали об этом в Spark 2.0, API-интерфейсы DataFrame будут объединены с API-интерфейсами наборов данных, объединяя возможности обработки данных в библиотеках. Благодаря этой унификации у разработчиков теперь меньше концепций, которые нужно изучать или запоминать, и они работают с одним высокоуровневым и типобезопасным API под названием Dataset.

Наборы данных

Начиная с Spark 2.0, набор данных имеет две различные характеристики API: строго типизированный API и нетипизированный API , как показано в таблице ниже.По идее, DataFrame можно рассматривать как псевдоним для коллекции универсальных объектов Dataset [Row] , где Row - это общий нетипизированный JVM-объект. Набор данных, напротив, представляет собой набор из строго типизированных объектов JVM, продиктованных классом case, который вы определяете в Scala, или классом в Java.

Типизированные и нетипизированные API

Примечание: Поскольку Python и R не имеют безопасности типов во время компиляции, у нас есть только нетипизированные API, а именно DataFrames.

Преимущества API наборов данных

Как разработчик Spark, вы получаете выгоду от унифицированных API DataFrame и Dataset в Spark 2.0 несколькими способами.

1. Статическая типизация и безопасность типов во время выполнения

Рассматривайте статическую типизацию и безопасность во время выполнения как спектр, при этом SQL наименее ограничивает набор данных, наиболее ограничивая его. Например, в строковых запросах Spark SQL вы не узнаете синтаксическую ошибку до момента выполнения (что может быть дорогостоящим), тогда как в DataFrames и Datasets вы можете обнаруживать ошибки во время компиляции (что экономит время разработчика и затраты).То есть, если вы вызываете функцию в DataFrame, которая не является частью API, компилятор ее поймает. Однако он не обнаружит несуществующее имя столбца до выполнения.

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

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

2. Абстракция высокого уровня и настраиваемое представление структурированных и полуструктурированных данных

DataFrames как набор наборов данных [строка] отображает структурированное настраиваемое представление в ваши полуструктурированные данные. Например, допустим, у вас есть огромный набор данных событий устройства Интернета вещей, представленный в формате JSON.Поскольку JSON является частично структурированным форматом, он хорошо подходит для использования набора данных как коллекции строго типизированного набора данных [DeviceIoTData] .

 {"device_id": 198164, "device_name": "sensor-pad-198164owomcJZ", "ip": "80.55.20.25", "cca2": "PL", "cca3": "POL", "cn": «Польша», «широта»: 53.080000, «долгота»: 18.620000, «масштаб»: «Цельсий», «темп»: 21, «влажность»: 65, «уровень батареи»: 8, «c02_level»: 1408, «ЖКД». ":" красный "," отметка времени ": 1458081226051} 

Каждую запись JSON можно выразить как DeviceIoTData , настраиваемый объект, с классом сценария Scala.

 case class DeviceIoTData (battery_level: Long, c02_level: Long, cca2: String, cca3: String, cn: String, device_id: Long, device_name: String, влажность: Long, ip: String, latitude: Double, lcd: String, longitude : Double, масштаб: String, temp: Long, отметка времени: Long) 

Затем мы можем прочитать данные из файла JSON.

 // читаем файл json и создаем набор данных из
// класс случая DeviceIoTData
// ds теперь представляет собой набор объектов JVM Scala DeviceIoTData
val ds = искра.read.json («/ databricks-public-datasets / data / iot / iot_devices.json»). как [DeviceIoTData] 

В приведенном выше коде под капотом происходят три вещи:

1. Spark считывает JSON, определяет схему и создает коллекцию DataFrames.
2. На этом этапе Spark преобразует ваши данные в DataFrame = Dataset [Row] , коллекцию универсального объекта Row, поскольку ему неизвестен точный тип.
3. Теперь Spark преобразует набор данных [строка] -> набор данных [DeviceIoTData] зависящий от типа объект Scala JVM в соответствии с классом DeviceIoTData .

Большинство из нас, кто работает со структурированными данными, привыкли просматривать и обрабатывать данные в виде столбцов или доступа к определенным атрибутам внутри объекта. Используя Dataset как коллекцию из типизированных объектов Dataset [ElementType] , вы легко получаете как безопасность во время компиляции, так и настраиваемое представление для строго типизированных объектов JVM. И полученный в результате строго типизированный Dataset [T] из приведенного выше кода можно легко отобразить или обработать с помощью высокоуровневых методов.

3. Простота использования API со структурой

Хотя структура может ограничивать контроль над тем, что ваша программа Spark может делать с данными, она вводит богатую семантику и простой набор операций, специфичных для предметной области, которые могут быть выражены в виде высокоуровневых конструкций. Однако большинство вычислений можно выполнить с помощью высокоуровневых API Dataset. Например, гораздо проще выполнить операции agg , select , sum , avg , map , filter или groupBy путем доступа к строкам DeviceIoTData типизированного объекта Dataset, чем использование RDD поля данных.

Выражение ваших вычислений в API, зависящем от предметной области, намного проще и легче, чем с выражениями типа алгебры отношений (в RDD). Например, приведенный ниже код filter () и map () создаст еще один неизменяемый набор данных.

 // Используем filter (), map (), groupBy () country и вычисляем avg ()
// для температуры и влажности. Эта операция приводит к
// еще один неизменяемый набор данных. Запрос проще читать,
// и выразительный

val dsAvgTmp = ds.filter (d => {d.temp> 25}). map (d => (d.temp, d.humidity, d.cca3)). groupBy ($ "_ 3"). avg ()

// отображаем результирующий набор данных
дисплей (dsAvgTmp) 

4. Производительность и оптимизация

Наряду со всеми вышеперечисленными преимуществами нельзя упускать из виду повышение эффективности использования пространства и повышения производительности при использовании DataFrames и API-интерфейсов Dataset по двум причинам.

Во-первых, поскольку API-интерфейсы DataFrame и Dataset построены на основе механизма Spark SQL, он использует Catalyst для создания оптимизированного логического и физического плана запроса.В API-интерфейсах R, Java, Scala или Python DataFrame / Dataset для всех запросов типа отношения применяется один и тот же оптимизатор кода, что обеспечивает эффективное пространство и скорость. В то время как типизированный API Dataset [T] оптимизирован для задач инженерии данных, нетипизированный Dataset [Row] (псевдоним DataFrame) работает еще быстрее и подходит для интерактивного анализа.

Во-вторых, поскольку Spark как компилятор понимает ваш объект JVM типа Dataset, он сопоставляет ваш объект JVM конкретного типа с представлением внутренней памяти Tungsten с помощью кодировщиков.В результате Tungsten Encoders могут эффективно сериализовать / десериализовать объекты JVM, а также генерировать компактный байт-код, который может выполняться с превосходной скоростью.

Когда мне следует использовать DataFrames или Datasets?

• Если вам нужна расширенная семантика, высокоуровневые абстракции и специфичные для домена API, используйте DataFrame или Dataset.
• Если ваша обработка требует выражений высокого уровня, фильтров, карт, агрегации, средних значений, суммы, SQL-запросов, столбцового доступа и использования лямбда-функций для полуструктурированных данных, используйте DataFrame или Dataset.
• Если вы хотите более высокий уровень безопасности типов во время компиляции, хотите типизированные объекты JVM, воспользоваться преимуществами оптимизации Catalyst и эффективно генерировать код Tungsten, используйте Dataset.
• Если вам нужна унификация и упрощение API-интерфейсов в библиотеках Spark, используйте DataFrame или Dataset.
• Если вы являетесь пользователем R, используйте DataFrames.
• Если вы пользователь Python, используйте DataFrames и вернитесь к RDD, если вам нужно больше контроля.

Обратите внимание, что вы всегда можете легко взаимодействовать или конвертировать из DataFrame и / или Dataset в RDD простым вызовом метода .rdd . Например,

 // выбираем определенные поля из набора данных, применяем предикат
// используя метод where (), конвертируем в RDD и показываем первые 10
// RDD строки
val deviceEventsDS = ds.select ($ "имя_устройства", $ "cca3", $ "c02_level"). где ($ "c02_level"> 1300)
// конвертируем в RDD и берем первые 10 строк
val eventsRDD = deviceEventsDS.rdd.take (10) 

Собираем все вместе

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

Изучив уроки, которые мы извлекли из ранних выпусков Spark - как упростить Spark для разработчиков, как оптимизировать и сделать его производительным - мы решили поднять низкоуровневые API RDD до высокоуровневой абстракции как DataFrame и Dataset и для построения этой унифицированной абстракции данных между библиотеками поверх оптимизатора Catalyst и Tungsten.

Выберите один - DataFrames и / или Dataset или RDD API - который соответствует вашим потребностям и сценарию использования, но я не удивлюсь, если вы попадете в лагерь большинства разработчиков, которые работают со структурными и полуструктурированными данными.

Что дальше?

Вы можете попробовать Apache Spark 2.2 на Databricks.

Вы также можете посмотреть презентацию Spark Summit «История трех API-интерфейсов Apache Spark: RDD против DataFrames и наборов данных»

Если вы еще не зарегистрировались, попробуйте Databricks прямо сейчас.

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

Spark и Spark Streaming Unit Testing

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

Модульные или интеграционные тесты, вот в чем вопрос

Наше гипотетическое приложение Spark извлекает данные из Apache Kafka, применяет преобразования, используя RDD а также DStreams и сохранять результаты в базе данных Cassandra или Elastic Search.На производстве приложение Spark разворачивается в кластере YARN или Mesos, и все склеивается с помощью ZooKeeper. Общая картина архитектуры потоковой обработки представлена ​​ниже:

Множество движущихся частей, не так-то просто настроить и протестировать. Даже с автоматической подготовкой, реализованной с помощью Vagrant, Docker и Ansible. Если вы не можете протестировать все, протестируйте хотя бы самую важную часть вашего приложения - преобразования, - реализованные с помощью Spark.

Spark утверждает, что он удобен для модульного тестирования с любой популярной средой модульного тестирования.Строго говоря, Spark поддерживает довольно легкое интеграционное тестирование, а не модульное тестирование, IMHO. Но все же гораздо удобнее тестировать логику трансформации локально, чем развертывать все части на YARN.

Существует запрос на вытягивание SPARK-1751, который добавляет поддержку «модульных тестов» для потоков Apache Kafka. Стоит ли нам идти по этому пути? Требуются встроенный ZooKeeper и встроенный Apache Kafka, тестовое устройство сложное и громоздкое. Возможно, тесты будут хрупкими и трудными в обслуживании. Такой подход имеет смысл для основной команды Spark, они хотят протестировать интеграцию Spark и Kafka.

Что нужно тестировать?

Наша логика преобразования реализована с помощью Spark, не более того. Но как проверить логику, так тесно связанную с Spark API (RDD, DStream)? Давайте определим, как организовано типичное приложение Spark. Наша гипотетическая структура приложения выглядит так:

1. Инициализировать SparkContext или StreamingContext .
2. Создать RDD или DStream для данного источника (например, Apache Kafka)
3. Оценить преобразования в RDD или DStream API.
4. Поместите результаты преобразования (например, агрегаты) во внешнюю базу данных.

Контекст

SparkContext и StreamingContext можно легко инициализировать для целей тестирования. Установите главный URL-адрес на , локальный , запустите операции, а затем корректно остановите контекст.

 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21 год
 

  class SparkExampleSpec расширяет FlatSpec с помощью BeforeAndAfter {

  частный мастер val = "местный [2]"
  private val appName = "пример-искра"

  частный var sc: SparkContext = _

  перед {
    val conf = новый SparkConf ()
      .setMaster (мастер)
      .setAppName (имя приложения)

    sc = новый SparkContext (conf)
  }

  после {
    if (sc! = null) {
      sc.stop ()
    }
  }
  (...)
  

 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21 год
22
23
24
25
26 год
27
28 год
 

  class SparkStreamingExampleSpec расширяет FlatSpec с помощью BeforeAndAfter {

  частный мастер val = "местный [2]"
  private val appName = "пример-искра-поток"
  private val batchDuration = Секунды (1)
  private val checkpointDir = Файлы.createTempDirectory (имя приложения) .toString

  частный var sc: SparkContext = _
  частный var ssc: StreamingContext = _

  перед {
    val conf = новый SparkConf ()
      .setMaster (мастер)
      .setAppName (имя приложения)

    ssc = новый StreamingContext (conf, batchDuration)
    ssc.checkpoint (checkpointDir)

    sc = ssc.sparkContext
  }

  после {
    if (ssc! = null) {
      ssc.stop ()
    }
  }

  (...)
  

RDD и DStream

Проблема в том, как создать RDD или DStream.В целях тестирования его необходимо упростить, чтобы избежать использования встроенных Kafka и ZooKeeper. Ниже вы можете найти примеры создания RDD и DStream в памяти.

 1
2
 

  val lines = Seq («Быть ​​или не быть.», «Вот в чем вопрос.»)
val rdd = sparkContext.parallelize (строки)
  

 1
2
3
4
5
 

  val lines = mutable.Queue [RDD [String]] ()
val dstream = streamingContext.queueStream (строки)

// добавляем данные в DStream
lines + = sparkContext.makeRDD (Seq («Быть ​​или не быть.», «Вот в чем вопрос.»))
  

Логика преобразования

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

 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
 

  case class WordCount (word: String, count: Int)

object WordCount {
  def count (строки: RDD [String], stopWords: Set [String]): RDD [WordCount] = {
    val слова = строки.flatMap (_. split ("\\ s"))
      .map (_. strip (","). strip ("."). toLowerCase)
      .filter (! stopWords.contains (_)). filter (! _. isEmpty)

    val wordCounts = words.map (word => (word, 1)). reduceByKey (_ + _). map {
      case (word: String, count: Int) => WordCount (word, count)
    }

    val sortedWordCounts = wordCounts.sortBy (_. слово)

    sortedWordCounts
  }
}
  

Искровой тест

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

 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
 

  «Самая известная цитата Шекспира» должна быть «засчитана» в {
    Дано ("цитата")
    val lines = Array ("Быть или не быть.", "Вот в чем вопрос.")

    Дано ("стоп-слова")
    val stopWords = Set ("the")

    Когда ("считать слова")
    val wordCounts = WordCount.count (sc.parallelize (строки), стоп-слова) .collect ()

    Затем («подсчитано слов»)
    wordCounts должно быть равно (Array (
      WordCount ("быть"; 2),
      WordCount ("есть"; 1),
      WordCount ("не"; 1),
      WordCount ("или", 1),
      WordCount ("вопрос", 1),
      WordCount ("то", 1),
      WordCount ("к", 2)))
  }
  

Тест Spark Streaming

Spark Streaming намного сложнее тестировать. Полный контроль над часами необходим для ручного управления пакетами, слайдами и окнами.Без контролируемых часов вы бы закончили сложные тесты с множеством вызовов Thread.sleeep . И выполнение теста займет много времени. Единственный минус - у вас не будет лишнего времени на кофе во время выполнения тестов.

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

 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
 

  пакет org.apache.spark.streaming

импортировать org.apache.spark.streaming.util.ManualClock

class ClockWrapper (ssc: StreamingContext) {

  def getTimeMillis (): Long = manualClock (). currentTime ()

  def setTime (timeToSet: Long) = manualClock (). setTime (timeToSet)

  def advance (timeToAdd: Long) = manualClock (). addToTime (timeToAdd)

  def waitTillTime (targetTime: Long): Long = manualClock ().waitTillTime (targetTime)

  private def manualClock (): ManualClock = {
    ssc.scheduler.clock.asInstanceOf [ManualClock]
  }

}
  

Теперь тест Spark Streaming можно эффективно реализовать. Тест не должен ждать системных часов, и тест выполняется с точностью до миллисекунды. Вы можете легко протестировать свой оконный сценарий от самого начала до самого конца. С данной структурой \ when \ then вы сможете понять протестированную логику без дополнительных объяснений.

 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21 год
22
23
24
25
26 год
27
28 год
29
30
31 год
32
33
34
35 год
36
37
38
39
40
41 год
42
43 год
44 год
45
46
47
48
49
50
51
52
53
54
55
56
57
 

  «Набор образцов» следует «засчитать» в {
  Дано ("контекст потоковой передачи инициализирован")
  val lines = mutable.Queue [RDD [String]] ()

  var results = ListBuffer.empty [массив [WordCount]]

  WordCount.count (ssc.queueStream (строки), windowDuration, slideDuration) {(wordsCount: RDD [WordCount], time: Time) =>
    результаты + = wordsCount.собирать()
  }

  ssc.start ()

  Когда ("первый набор слов в очереди")
  lines + = sc.makeRDD (Seq ("a", "b"))

  Затем («количество слов после первого слайда»)
  clock.advance (slideDuration.milliseconds)
  в конце концов (тайм-аут (1 секунда)) {
    results.last должен быть равен (Array (
      WordCount ("а", 1),
      WordCount ("b", 1)))
  }

  Когда ("второй набор слов в очереди")
  lines + = sc.makeRDD (Seq ("b", "c"))

  Затем («количество слов после второго слайда»)
  clock.advance (slideDuration.milliseconds)
  в конце концов (тайм-аут (1 секунда)) {
    полученные результаты.последний должен быть равен (Массив (
      WordCount ("а", 1),
      WordCount ("b"; 2),
      WordCount ("c", 1)))
  }

  Когда ("больше ничего не стоит в очереди")

  Затем («слово считается после третьего слайда»)
  clock.advance (slideDuration.milliseconds)
  в конце концов (тайм-аут (1 секунда)) {
    results.last должен быть равен (Array (
      WordCount ("a", 0),
      WordCount ("b", 1),
      WordCount ("c", 1)))
  }

  Когда ("больше ничего не стоит в очереди")

  Затем («слово считается после четвертого слайда»)
  clock.advance (slideDuration.milliseconds)
  в конце концов (тайм-аут (1 секунда)) {
    полученные результаты.последний должен быть равен (Массив (
      WordCount ("a", 0),
      WordCount ("b", 0),
      WordCount ("c", 0)))
  }
}
  

Один комментарий к В конечном итоге использование признака . Эта черта необходима, потому что Spark Streaming является многопоточным приложением, и результаты не вычисляются немедленно. Я обнаружил, что для Spark Streaming достаточно тайм-аута в 1 секунду для расчета результатов. Тайм-аут не связан с длительностью пакета, слайда или окна.

Сводка

Полный рабочий проект опубликован на GitHub.