🚀 автор : «Большие данные, маленький дзен» 🚀Введение в статью：Эта статья принадлежитSparkСерия статей,В колонке будет записываться контент от основ Spark до продвинутых. 🚀 Контент предполагает построение кластера начального уровня Spark.,основные компоненты,RDD,Использование оператора,основополагающие принципы,SparkCore,SparkSQL,SparkStreaming и т. д.,Адрес столбца Spark.Добро пожаловать, друзья, подписывайтесь💪

Обычно используетсяоператор Коллекция

• Обзор операторов в Spark
• Разница и связь между оператором преобразования и оператором действия
• Краткое описание распространенных операторов преобразования
  • оператор карты
  • оператор FlatMap
  • оператор фильтра
  • Оператор MapPartitions
  • Оператор уменьшения по ключу
  • Оператор groupByKey
  • оператор выборки
  • оператор сортировки по
  • отдельный оператор
  • оператор объединения
  • ОператорfoldByKey
  • оператор вычитания
  • присоединиться к оператору
• Краткое описание операторов общего действия
  • оператор сокращения
  • оператор кольцета
  • оператор счета
  • взять оператор
  • оператор foreach

Обзор операторов в Spark

Операторы в RDD функционально делятся на две категории:

• 1.Transformation(Конвертироватьоператор) это будет в уже существующем RDD Создайте новый на РДР, который также создает кровное родство и связи между РДР.
• 2.Action(действиеоператор) Выполнять вычислительные задачи для каждого раздела, Результат возвращается в Driver середина

Функции

• 1.Spark серединаall Transformations да инерция из, Результаты будут получены не сразу. Будут записаны только операции, которые будут применены к набору данных. Когда необходимо будет вернуть результаты. Driver час, выполнит эти операции,Эта функция называется инерция Оценивать
• 2. При запуске каждого Действия все связанные СДР Трансформации будут пересчитаны.

Разница и связь между оператором преобразования и оператором действия

Оператор преобразования — это операция в Spark, которая используется для преобразования одного RDD в другой. Его можно использовать для создания нового RDD или для преобразования существующего RDD. Они предоставляют общий метод для завершения преобразования RDD, такой как карта, фильтр, groupByKey и т. д.

Операторы действий — это еще один тип операций в Spark. Они используются для сбора данных из RDD или вычисления результатов из RDD, например сбор, сокращение, подсчет и т. д. Операторы действий могут быть рассчитаны на основе результатов операторов преобразования СДР или на основе набора СДР.

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

Краткое описание распространенных операторов преобразования

оператор карты

• Карта преобразует данные RDD в другие формы во взаимосвязи «один к одному». Входной и выходной разделы являются взаимно однозначными.
• collect: Соберите все элементы эластичного распределенного набора данных в массив середина для удобства наблюдения. Подходит для небольших данных
• take: вывести отображаемый номер соответствующих данных
• foreach(println(_)): переход для просмотра данных
• Результат: 1,4,9,16 (йо,1) (пай,1) (хс,1)

def mapTest(): Unit ={

    val value = sc.parallelize(List(1, 2, 3, 4)).map(
      value=>value*2
    ).collect().foreach(println(_))
    val works = sc.parallelize(List("yo", "pai", "xc")).map(
      work => (work, 1)
    ).collect().take(2).foreach(println(_))
  }

оператор FlatMap

• оператор Функция FlatMap — разделить строку данных на несколько элементов.,и поместить все элементы в новую коллекцию середина,Верните новый RDD.
• оно и оператор Отличие карт в том, что оператор карты Разделить строку данных на один элемент,и выложи в новую коллекцию середина,
• А оператор FlatMap может разбить строку данных на несколько элементов.,и поместить все элементы в новую коллекцию середина。
• Результат: y o p a i x c’’

@Test
  def flatmapTest(): Unit ={
    val works = sc.parallelize(List("yo", "pai", "xc")).flatMap(
      work=>(work)
    ).collect().foreach(println(_))
  }

оператор фильтра

• Фильтр оператора sparkсередина используется для применения функции к каждому элементу RDDсередина.,В зависимости от возвращаемого значения функции даtrue или даfalse решается, помещать ли элемент в новый RDDсередина.
• То есть, да, оператор фильтра может быть основан на логике пользовательской функции середина,Отфильтровать новый RDD из источника RDDсередина.
• Результат: пай хс

@Test
  def filterTest(): Unit ={
    val works = sc.parallelize(List("yo", "pai", "xc")).filter(
      //Удаляем те, которые не содержат поля yo
      work=>(!work.contains("yo"))
    ).collect().foreach(println(_))
  }

Оператор MapPartitions

• оператор карты Операция «один на один»,Сопоставит каждый элемент одного RDDсередина с другим RDDсередина;
• И Оператор MapPartitionsда операция один-ко-многим,Он сопоставит каждый раздел одного RDDсередина с другим RDDсередина.,Элементы каждого раздела середина будут обработаны сразу,Уменьшите количество операций,Повышенная эффективность обработки.
• MapPartitions — то же самое, что и оператор карты,Просто мапда преобразует каждый фрагмент данных,MapPartitions преобразует весь раздел
• Сценарий:
• 1. Если за картой стоит оператор доступа к базе данных, то при наличии десятков тысяч запрашиваемых данных необходимо производить десятки тысяч установлений соединения. Это явно нелогично.
• 2. MapPartitions(foreachPartition) работает с итератором каждого раздела rddсередина. Если вам необходимо часто создавать дополнительные объекты в процессе отображения (например, записывать данные rddсередина через jdbc в базу данных
• Map необходимо создать ссылку для каждого элемента, а MapPartition создает ссылку для каждого раздела), тогда MapPartitions намного эффективнее, чем Map.
• Результат: 10,20,30,40,50

@Test
  def mapPartitionsTest(): Unit ={
    val works = sc.parallelize(List(1,2,3,4,5)).mapPartitions(
      //Он получает функцию, а функция получает набор. Для передачи набора требуется возврат набора.
      //Функция требует получения набора, преобразования каждого фрагмента данных в Iterator[T] в набор, а затем возврата набора.
      //Iterator[T] => Iterator[U]
     item=>{
       val ele=item.map(item=>item*10)
       ele
     }
    )
    println(works.collect().mkString(","))
  }

Оператор уменьшения по ключу

• reduceByKey((V, V) ⇒ V, numPartition)
• Используется для объединения одинаковых элементов парыRDD.
• Его функция — выполнять операции сокращения над элементами каждого ключа парыRDDсередина.,Агрегируйте значения значений, соответствующие ключу, вместе,Тем самым реализуя операцию агрегации на пареRDD.
• Результат: (Брат Юн, 198 лет) (Сяо Мин, 97 лет)

@Test
  def reduceByKeyTest(): Unit ={
    val works = sc.parallelize(Seq(("Брат Юн", 100), («Брат Юн», 98), («Сяо Мин», 97))).reduceByKey(
      (a,b)=>a+b
    )
    println(works.collect().foreach(println(_)))
  }

Оператор groupByKey

• Важная операция преобразования groupByKeyдаSparkсередина.,Его функция — группировать элементы, соответствующие каждой клавише.,Затем верните сгруппированные результаты в форме ключ-значение.,
• Это оригинальный ключ серединакейда,итератор значения,Итератор середина хранит все элементы, соответствующие ключу.
• Оператор groupByKey может использоваться для группировки элементов RDDсередина.,Иногда его также можно использовать для операций агрегирования.,Но его производительность намного ниже, чем у других агрегатных функций.,Поэтому его использование обычно не рекомендуется.
• результат:
• (Брат Юн,CompactBuffer(100))
• (Сяохун,CompactBuffer(98))
• (Сяо Мин, CompactBuffer(97, 77))

//Создаем RDD из локальной коллекции
    val rdd = sc.parallelize(Seq(("Брат Юн", 100), («Красный», 98), («Сяо Мин», 97), («Сяо Мин», 77))).groupByKey()
    println(rdd.collect().foreach(println(_)))
  }

оператор выборки

• sample(withReplacement, fraction, seed)
• оператор выборакидаспарксередина используется для выборки из RDDсередина,Он может извлечь часть образца из RDDсередина в соответствии с указанной пропорцией или количеством.,Его можно использовать для исследования данных, разработки моделей и т. д.

@Test
  def sampleTest(){
    //Создаем RDD из локальной коллекции
    val rdd = sc.parallelize(List(1,2,3,4,5,6,7)).sample(
      //Случайным образом рисуем 3 числа
      withReplacement = true,2
    )
    println(rdd.collect().foreach(println(_)))
  }

оператор сортировки по

• оператор сортировки пода Сортировать элементы RDDсередина по заданным правилам.,Его возвращаемый тип — отсортированный RDD.
• результат: (Bob,70) (John,80) (Tom,90)

@Test
  def sortByTest(){
    val rdd = sc.parallelize(Array(("Tom",90),("Bob",70),("John",80))).sortBy(_._2)
    println(rdd.collect().mkString(" "))
  }

отдельный оператор

• Different удаляет повторяющиеся элементы из RDDсередина.
• результат: 4 6 2 1 3 5

@Test
  def distinctTest(){
    val rdd = sc.parallelize(List(1,2,3,4,4,5,6,6))
    val distinctRDD = rdd.distinct()
    println(distinctRDD.collect().mkString(" "))
  }

оператор объединения

• оператор объединение даспарксередина оператор для объединения нескольких СДР в один СДР,Результат RDDсередина содержит все элементы входного RDDсередина.,И не сбрасывать вес.
• Команда subtract может вычесть элементы одного RDDсередина из другого RDD, чтобы получить новый RDD.
• результат: 1 2 3 1 2 3

@Test
  def unionTest(){
    val rdd1 = sc.parallelize(List(1,2,3))
    val rdd2 = sc.parallelize(List(1,2,3))
    val rdd = rdd1.union(rdd2)
    println(rdd.collect().mkString(" "))
  }

ОператорfoldByKey

• foldByKey(zeroValue)((V, V) ⇒ V)
• оператор выполняет операции агрегирования данных типа «ключ-значение» RDDсередина по ключу,Агрегировать значения, соответствующие каждому ключу,
• Объедините агрегированный результат с нулевым значением, чтобы вернуть новый СДР.
• Каждый элемент нового середина RDD имеет пару ключ-значение, и его серединаkeyда является ключом исходного середина RDD, а значение даzeroValue является результатом агрегирования значения, соответствующего исходному середина RDD.
• результат: (Брат Юн, 21 год) (Сяо Мин, 22 года)

@Test
  def foldByKeyTest(){
    //Создаем RDD из локальной коллекции
    val rdd = sc.parallelize(Seq(("Брат Юн", 1), («Сяо Мин», 1), («Сяо Мин», 1))).foldByKey(zeroValue = 20)((a,b)=>(a+b))
    println(rdd.collect().foreach(println(_)))
  }

оператор вычитания

• оператор Операция RDD для вычитанияdasparkсередина,Он может получать два RDD в качестве параметров.,и возвращает новый RDD
• Новая середина RDD содержит элементы, которые существуют в первой середине RDD, но не существуют во второй середине RDD.
• результат: 1 2

@Test
  def subtractTest(){
    val rdd1 = sc.parallelize(Seq(1,2,3,4,5))
    val rdd2 = sc.parallelize(Seq(3,4,5,6,7))

    val rdd3 = rdd1.subtract(rdd2)
    println(rdd3.collect().foreach(println(_)))
  }

присоединиться к оператору

• присоединиться к операторудаsparkсерединавродевнутреннее соединениеоператор,Он объединяет кортежи с одним и тем же ключом из двух наборов данных. Его можно использовать между RDD, DataFrame и Dataset.,
• Его серединаRDD и DataFrame можно соединить с помощью соединения с оператором, а Dataset — с помощью joinWithoperator.
• результат: (2,(xc,xc1)) (1,(yo,yo1)) (3,(yong,yong1))

@Test
  def joinTest(){
    val rdd1 = sc.makeRDD(Array((1, "yo"), (2, "xc"), (3, "yong")))
    val rdd2= sc.makeRDD(Array((1, "yo1"), (2, "xc1"), (3, "yong1")))
    val rdd3=rdd1.join(rdd2)
    println(rdd3.collect().mkString(" "))
  }

Краткое описание операторов общего действия

оператор сокращения

• сокращение сначала объединяет данные внутри разделов, затем объединяет данные между разделами
• результат:10

@Test
  def reduceTest(){
    //Создаем RDD из локальной коллекции
    val rdd = sc.parallelize(List(1,2,3,4)).reduce(_+_)
    println(rdd)
  }

оператор кольцета

• colcet сначала возвращает набор результирующих данных в виде массива Array
• результат:10 1 2 3 4

@Test
  def collcetTest(){
    //Создаем RDD из локальной коллекции
    val rdd = sc.parallelize(List(1,2,3,4))
    println(rdd.collect().mkString(" "))
  }

оператор счета

• count возвращает количество элементов в RDD
• результат: 4

 @Test
  def countTest(){
    //Создаем RDD из локальной коллекции
    val rdd = sc.parallelize(List(1,2,3,4))
    println(rdd.count())
  }

взять оператор

• take возвращает массив, состоящий из первых n элементов RDD
• результат: 1 2

@Test
  def takeTest(){
    //Создаем RDD из локальной коллекции
    val rdd: RDD[Int] = sc.parallelize(List(1,2,3,4))
    println(rdd.take(2).mkString(" "))
  }

оператор foreach

• foreach обходит элементы RDDсередина
• результат: 1 2

  @Test
  def foreachTest(){
    //Создаем RDD из локальной коллекции
    val rdd: RDD[Int] = sc.parallelize(List(1,2,3,4))
    println(rdd.take(2).foreach(println(_)))
  }

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