Сортировка массива сложного типа данных в Spark

Мы можем использовать функцию sort() или функцию orderBy() для сортировки массива Spark, но эти функции могут не работать, если массив имеет сложный тип данных. Для таких массивов сложных типов данных нам необходимо использовать разные способы сортировки массива сложного типа данных в PySpark, который будет определен в этой статье с использованием Python.

Способы сортировки массива сложных типов данных

Ниже приведены способы сортировки массива сложных типов данных в PySpark:

Использование функции выражение()
Использование функции agg()

Использование функции выражение():

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

Syntax: expr(str)

Здесь,

str: определяет строковый аргумент, который определяет тип данных столбца Spark.

Пример. В этом примере мы использовали три столбца: класс, раздел и name_age, как показано на изображении. Здесь name_age — это столбец со сложным типом данных, который мы будем сортировать по возрасту в порядке убывания.

Python3

# Python program to sort an array of a complex

# data type in Spark using expr() function

import pyspark

from pyspark.sql import SparkSession

from pyspark.sql.functions import expr

# Create a Spark Session

spark=SparkSession.builder.getOrCreate()

# Create an array

data = [

    [3, 'A', [["Ishita", 7], ["Ashish", 9], ["Ramesh", 10], ["Mathew", 8]]],

    [7, 'D', [["Vinayak", 13], ["Pranjal", 14], ["Isha", 12], ["Adarsh", 11]]],

]

# Create a Spark data frame

df = spark.createDataFrame(

    data, "class:int, section:string, name_age:array<struct<name:string, age:int>>"

)

# Sort the name_age column by age in descending order

df = df.withColumn(

    "name_age",

    expr(

        "reverse(array_sort(transform(name_age,x->struct(x['age'] as age,x['name'] as name))))"

    ),

)

# Display the data frame

df.show(2,False)

Выход

Использование функции agg():

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

Syntax:

agg(*exprs)

Здесь,

exprs: определяет выражение, которое необходимо передать в функцию.

Пример. В этом примере мы использовали три столбца: раздел, имя и последовательность, как показано на изображении. Затем мы отсортировали имя столбца на основе последовательности после группировки в разделе столбцов.

Python3

# Python program to sort an array of a complex

# data type in Spark using agg() function

# Import the Pyspark library

from pyspark.sql import *

import pyspark

from pyspark.sql import SparkSession

# Create a Spark Session

spark=SparkSession.builder.getOrCreate()

# Create the Spark data frame

data_df = spark.createDataFrame([

  Row(section='A', name='Ashish', sequence=1),

  Row(section='B', name='Bharti', sequence=1),

  Row(section='B', name='Charlie', sequence=2),

  Row(section='A', name='Marie', sequence=2),

  Row(section='C', name='Prabhakar', sequence=1),

  Row(section='D', name='Shrey', sequence=1),

  Row(section='C', name='Rose', sequence=2),

  Row(section='B', name='Ishita', sequence=3),

  Row(section='C', name='Samarth', sequence=3),

  Row(section='A', name='Vinayak', sequence=4),

  Row(section='A', name='Pranjal', sequence=3),

])

# Display the Spark data frame

views_df = data_df.select('section', 'name', 'sequence')

# Create a local temporary view

views_df.createOrReplaceTempView('views')

# Sort array of complex data type

views_df= (views_df

  .groupBy('section')

  .agg(

    array_sort(

      collect_list(struct('sequence', 'name'))

    ).alias('sorted_names')

  ).select("section", col("sorted_names.name").alias("Sorted Names"))

).show(20, False)

Выход