СПАРК 2.2.2 - Объединение нескольких СДР, выдающих из памяти кроме себя. Результирующий RDD имеет 124 столбца. Каким должен быть оптимальный метод соединения?

Question

У меня есть файл, который имеет несколько значений для каждого номера телефона. например:

phone_no circle operator priority1 attribute1 attribute2 attribute3 priority2 attribute1 attribute2 attribute3 
123445   delhi  airtel   1.0        info1      info2      info3      1.1        info4      info5      info6
987654   bhopal idea     1.1        info1      info2      info3      1.4        info4      info5      info6
123445   delhi  airtel   1.3        info1      info2      info3      1.0        info4      info5      info6

Каков мой ожидаемый результат: для каждого номера телефона выберите минимальное значение P1 и соответствующие ему значения атрибутов.

Как показано в приведенном выше примере, для телефонного номера 123445 P1 в строке 1 меньше, чем P1 в строке 3 (1.0 <1.3), поэтому я хочу выбрать атрибуты 1,2 и 3 в строке 1 и как P2 в строке 3 имеет меньшее значение (1,0 <1,1), поэтому я хочу выбрать значения атрибута из строки 3. </p>

Вот что я хочу в табличном формате:

phone_no circle operator priority1 attribute1 attribute2 attribute3 priority2 attribute1 attribute2 attribute3 
123445   delhi  airtel   1.0        info1      info2      info3      1.0        info4      info5      info6
987654   bhopal idea     1.1        info1      info2      info3      1.4        info4      info5      info6

У меня есть 25 различных значений приоритета, и у каждого значения приоритета есть 4 различных атрибута, поэтому мои общие столбцы составляют около 125.

Что я пробовал до сих пор:

1. Создайте кадр данных, в котором в качестве ключа указан номер телефона и минимум каждого значения приоритета.
2. Создайте еще один Dataframe со значениями min (Priority1) и соответствующими атрибутами для каждого телефонного номера.
3. Создайте еще один Dataframe со значениями min (Priority2) и соответствующими атрибутами для каждого телефонного номера.
4. Соедините эти два кадра данных на номер телефона, чтобы получить полную информацию и сохранить этот кадр на диск.

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

РЕДАКТИРОВАТЬ 1: Вот ссылка на вставку из того, что я сделал: https://pastebin.com/ps4f1KSh

Answer 1

Я бы, наверное, использовал оконные функции:

from pyspark.sql.window import Window
import pyspark.sql.functions as spf

df = spark.createDataFrame([
    (123, 1, 'a', 2, 'c'),
    (123, 2, 'b', 1, 'd'),
    (456, 3, 'e', 4, 'f')
], ['phone', 'priority1', 'attribute1', 'priority2', 'attribute2'])

w = Window.partitionBy('phone')
df2 = (
    df
    .select(
        'phone',
        spf.first('attribute1').over(w.orderBy('priority1')).alias('attribute1'),
        spf.first('attribute2').over(w.orderBy('priority2')).alias('attribute2'),
    )
)

(
    df2
    .groupby('phone')
    .agg(*[spf.first(c).alias(c) for c in df2.columns if c != 'phone'])
    .toPandas()
)

Дает:

   phone attribute1 attribute2
0    123          a          d
1    456          e          f

Это упражнение для читателя, чтобы шаблонировать это (например, используя списки), чтобы обобщить все атрибуты и приоритеты.