Pyspark - выбор отличительных значений в столбце после groupby и orderBy - PullRequest
Новая
       5

Pyspark - выбор отличительных значений в столбце после groupby и orderBy

0 голосов
/ 09 июня 2018

поэтому моя таблица выглядит следующим образом: 

+-------------------+-------+----------+------------+
|            trip_id|line_id|  ef_ar_ts|     station|
+-------------------+-------+----------+------------+
|80:06____:17401:000|  17401|         0|Schaffhausen|
|80:06____:17402:000|  17402|1505278458|Schaffhausen|
|80:06____:17403:000|  17403|         0|Schaffhausen|
|80:06____:17406:000|  17406|1505282110|Schaffhausen|
|80:06____:17409:000|  17409|         0|Schaffhausen|
|80:06____:17410:000|  17410|1505285757|Schaffhausen|
|80:06____:17411:000|  17411|         0|Schaffhausen|
|80:06____:17416:000|  17416|1505292890|Schaffhausen|
|80:06____:17417:000|  17417|         0|Schaffhausen|
|80:06____:17418:000|  17418|1505296501|Schaffhausen|
|80:06____:17419:000|  17419|         0|Schaffhausen|
|80:06____:17420:000|  17420|1505300253|Schaffhausen|
|80:06____:17421:000|  17421|         0|Schaffhausen|
|80:06____:17422:000|  17422|1505303814|Schaffhausen|
|80:06____:17423:000|  17423|         0|Schaffhausen|
|80:06____:17425:000|  17425|         0|Schaffhausen|
|80:06____:17426:000|  17426|1505307355|Schaffhausen|
|80:06____:17427:000|  17427|         0|Schaffhausen|
|80:06____:17428:000|  17428|1505310983|Schaffhausen|
|80:06____:17429:000|  17429|         0|Schaffhausen|
+-------------------+-------+----------+------------+

Это набор данных о поездах, и я хочу сделать следующее:

Группировать line_id изпоезда, так что у меня есть все мои станции вместе с их линией;упорядочить их по ( ef_ar_ts ) в каждой из этих групп;затем получите SET station в их последовательном порядке: один список на line_id.Таким образом, я закажу свои станции и восстановлю всю линию.

То, что я до сих пор пробовал, было так: 

from pyspark.sql.functions import udf
@functions.udf
def keepline(df):
    """Keep lines splitted;"""
    firstline = data1.first().trip_id

dftemp = df.where(data1.trip_id==firstline)

data1 = data1.fillna({'ef_ar_ts':0})

dftemp = dftemp.orderBy('ef_ar_ts')



return mylist

data2 = data1.select('*').groupby(data1.line_id).agg(udfmyfunc)

Любая помощь?Заранее!

1 Ответ

0 голосов
/ 09 июня 2018

Мы можем сгруппировать line_id , собрать ef_ar_ts & station столбцы и упорядочить обе коллекции, используя UDF. Надеюсь, это поможет.

Так какВаш фрейм данных имеет те же значения в станции, я добавил две строки с dummystation для справки, 

+-------------------+-------+----------+-------------+
|            trip_id|line_id|  ef_ar_ts|      station|
+-------------------+-------+----------+-------------+
|80:06____:17401:000|  17401|         0| Schaffhausen|
|80:06____:17402:000|  17402|1505278458| Schaffhausen|
|80:06____:17403:000|  17403|         0| Schaffhausen|
......................................................
......................................................
|80:06____:17427:000|  17427|         0| Schaffhausen|
|80:06____:17428:000|  17428|1505310983| Schaffhausen|
|80:06____:17429:000|  17429|         0| Schaffhausen|
|80:06____:17429:000|  17401|1505278478|dummystation1|
|80:06____:17429:000|  17429|1505307355|dummystation2|
+-------------------+-------+----------+-------------+

## group and collect for each line id ##
df1 = df.groupby('line_id').agg(F.collect_list('ef_ar_ts').alias('ef_ar_ts'),F.collect_list('station').alias('station'))

+-------+---------------+-----------------------------+
|line_id|ef_ar_ts       |station                      |
+-------+---------------+-----------------------------+
|17419  |[0]            |[Schaffhausen]               |
|17420  |[1505300253]   |[Schaffhausen]               |
|17403  |[0]            |[Schaffhausen]               |
|17406  |[1505282110]   |[Schaffhausen]               |
|17428  |[1505310983]   |[Schaffhausen]               |
|17421  |[0]            |[Schaffhausen]               |
|17427  |[0]            |[Schaffhausen]               |
|17411  |[0]            |[Schaffhausen]               |
|17416  |[1505292890]   |[Schaffhausen]               |
|17429  |[0, 1505307355]|[Schaffhausen, dummystation2]|
|17401  |[0, 1505278478]|[Schaffhausen, dummystation1]|
|17423  |[0]            |[Schaffhausen]               |
|17417  |[0]            |[Schaffhausen]               |
|17402  |[1505278458]   |[Schaffhausen]               |
|17418  |[1505296501]   |[Schaffhausen]               |
|17425  |[0]            |[Schaffhausen]               |
|17409  |[0]            |[Schaffhausen]               |
|17422  |[1505303814]   |[Schaffhausen]               |
|17426  |[1505307355]   |[Schaffhausen]               |
|17410  |[1505285757]   |[Schaffhausen]               |
+-------+---------------+-----------------------------+

## an UDF for merge both collections and sort them ##
from operator import itemgetter
udf1 = F.udf(lambda x,y : [st[1] for st in sorted(zip(x,y),key=itemgetter(0))])
df1.select('line_id',udf1('ef_ar_ts','station').alias('stations')).show(truncate=False)

+-------+-----------------------------+
|line_id|stations                     |
+-------+-----------------------------+
|17419  |[Schaffhausen]               |
|17420  |[Schaffhausen]               |
|17403  |[Schaffhausen]               |
|17406  |[Schaffhausen]               |
|17428  |[Schaffhausen]               |
|17421  |[Schaffhausen]               |
|17427  |[Schaffhausen]               |
|17411  |[Schaffhausen]               |
|17416  |[Schaffhausen]               |
|17429  |[Schaffhausen, dummystation2]|
|17401  |[Schaffhausen, dummystation1]|
|17423  |[Schaffhausen]               |
|17417  |[Schaffhausen]               |
|17402  |[Schaffhausen]               |
|17418  |[Schaffhausen]               |
|17425  |[Schaffhausen]               |
|17409  |[Schaffhausen]               |
|17422  |[Schaffhausen]               |
|17426  |[Schaffhausen]               |
|17410  |[Schaffhausen]               |
+-------+-----------------------------+
...