Как написать функцию pyspark groupby agg, которая выполняет сравнение строк

Question

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

    record_id, colA, colB ....file
    10086 , 'haha', 'ok'  ....'csv1'
    10086 , 'haha', 'oh no' ....'csv2'
    10087, 'ok', 'good' ........'csv1'
    10087, 'ok', 'good' ........'csv2’

и мне нужно получить фрейм данных, как показано ниже:

    record_id, colA, colB ....file
    10086 , 1, 0  ….0
    10087, 1, 1 ……..0

идея в том, что я хочу загрузить два CSV в кадры данных, создатьобъединение и получить df_both, который содержит двойные записи записей из обоих csvs.теперь мне просто нужно вызвать

    df_both.groupBy("record_id").agg(string_comparison('colA'), string_comparison('colB') ...).show() 

, чтобы сгенерировать логическую матрицу.

Мне просто нужно выяснить, как написать функцию агрегирования string_comparison, которая проверяет две строки с одинаковым record_id по каждому полю, а затем возвращает строку, похожую на "record_id, 0, 1, 0, 1 .... "

это выполнимо с помощью RDD, но Python с фреймом данных с искрой, который работает намного быстрее, чем RDD, есть ли способ, которым я могу этого достичь?Спасибо!

Answer 1

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

record_id COLA    COLB  colA    colB
10086     haha    ok    haha    oh no
10087     ok      good  ok      good

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

@udf
def compareCols(col1,col2):
    if col1==col2:
        return 1
    else:
        return 0
outputDF = AboveDF.withColumn('compareA', compareCols(col('COLA'),col('colA'))\
                  .withColumn('compareB', compareCols(col('COLB'),col('colB'))