Объединить строки данных pyspark на основе одного идентичного идентификатора - PullRequest
Новая
       10

Объединить строки данных pyspark на основе одного идентичного идентификатора

0 голосов
/ 10 января 2019

Я хочу сгладить набор данных hbase. Человек имеет 2 типа данных в другой таблице, почте и телефоне. После присоединения к ним rel_id = data_id я получаю 2 отдельные записи для одного человека. 

    |individual_id|rel_id|data_id|       mail|phone|role
    ---------------------------------------------------------
    |            1|   100|    100|a@gmail.com| null|secondary
    |            1|   200|    200|       null|  123|primary
    |            2|   300|    300|       null|  345|secondary
    |            2|   400|    400|b@gmail.com| null|primary
    ------------------------------------------------

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

    |individual_id|       mail|phone|mail_role |phone_role
    ------------------------------------------------------
    |            1|a@gmail.com|  123| secondary|primary
    |            2|b@gmail.com|  345|   primary|secondary
    -----------------------------------------------------

Буду очень признателен за любую помощь.

1 Ответ

0 голосов
/ 10 января 2019

Вы можете группировать по individual_id и использовать pyspark.sql.functions.first в качестве функции агрегирования. Вам нужно будет установить для параметра ignorenulls для first значение True: 

from pyspark.sql.functions import first

df.groupBy("individual_id").agg(
    first("mail", ignorenulls=True).alias("mail"), 
    first("phone", ignorenulls=True).alias("phone")
).show()
#+-------------+-----------+-----+
#|individual_id|       mail|phone|
#+-------------+-----------+-----+
#|            1|a@gmail.com|  123|
#|            2|b@gmail.com|  345|
#+-------------+-----------+-----+

Для обновленного вопроса вы можете использовать pyspark.sql.functions.when: 

from pyspark.sql.functions import when, col

df.groupBy("individual_id").agg(
    first("mail", ignorenulls=True).alias("mail"), 
    first("phone", ignorenulls=True).alias("phone"),
    first(when(col("mail").isNotNull(), col("role")), ignorenulls=True).alias("mail_role"),
    first(when(col("phone").isNotNull(), col("role")), ignorenulls=True).alias("phone_role"),
).show()
#+-------------+-----------+-----+---------+----------+
#|individual_id|       mail|phone|mail_role|phone_role|
#+-------------+-----------+-----+---------+----------+
#|            1|a@gmail.com|  123| seconary|   primary|
#|            2|b@gmail.com|  345|  primary| secondary|
#+-------------+-----------+-----+---------+----------+
...