Spark-фильтр нескольких групп строк в одну строку - PullRequest
Новая
       8

Spark-фильтр нескольких групп строк в одну строку

0 голосов
/ 15 мая 2018

Я пытаюсь получить следующее,

Допустим, у меня есть кадр данных со следующими столбцами 

id  | name  | alias
-------------------
1   | abc   | short
1   | abc   | ailas-long-1
1   | abc   | another-long-alias
2   | xyz   | short_alias
2   | xyz   | same_length
3   | def   | alias_1

Я хочу сгруппировать идентификатор и имя и выбрать более короткий псевдоним,

Ожидаемый вывод: 

id  | name  | alias
-------------------
1   | abc   | short
2   | xyz   | short_alias
3   | def   | alias_1

Я могу добиться этого, используя window и row_number, есть ли другой эффективный метод для получения того же результата.В общем случае условие фильтра третьего столбца может быть любым, в данном случае это длина поля.

Любая помощь будет принята с благодарностью.

Спасибо.

Ответы [ 2 ]

0 голосов
/ 15 мая 2018

Решение без окна (Не очень красиво ..) и самое простое, на мой взгляд, решение rdd: 

from pyspark.sql import functions as F
from pyspark.sql import HiveContext
hiveCtx = HiveContext(sc)

rdd = sc.parallelize([(1   , "abc"   , "short-alias"),
                     (1   , "abc"   , "short"),
                         (1   , "abc"   , "ailas-long-1"),
                         (1   , "abc"   , "another-long-alias"),
                         (2   , "xyz"   , "same_length"),
                         (2   , "xyz"   , "same_length1"),
                         (3   , "def"   , "short_alias") ])

df = hiveCtx.createDataFrame(\
rdd, ["id", "name", "alias"])

len_df = df.groupBy(["id", "name"]).agg(F.min(F.length("alias")).alias("alias_len"))

df = df.withColumn("alias_len", F.length("alias"))

cond = ["alias_len", "id", "name"]

df.join(len_df, cond).show()

print rdd.map(lambda x: ((x[0], x[1]), x[2]))\
    .reduceByKey(lambda x,y: x if len(x) < len(y) else y ).collect()

Выход: 

+---------+---+----+-----------+
|alias_len| id|name|      alias|
+---------+---+----+-----------+
|       11|  3| def|short_alias|
|       11|  2| xyz|same_length|
|        5|  1| abc|      short|
+---------+---+----+-----------+

[((2, 'xyz'), 'same_length'), ((3, 'def'), 'short_alias'), ((1, 'abc'), 'short')]
0 голосов
/ 15 мая 2018

Все, что вам нужно сделать, это использовать встроенную функцию length и использовать ее в функции window как 

from pyspark.sql import functions as f
from pyspark.sql import Window

windowSpec = Window.partitionBy('id', 'name').orderBy('length')

df.withColumn('length', f.length('alias'))\
    .withColumn('length', f.row_number().over(windowSpec))\
    .filter(f.col('length') == 1)\
    .drop('length')\
    .show(truncate=False)

, который должен дать вам 

+---+----+-----------+
|id |name|alias      |
+---+----+-----------+
|3  |def |alias_1    |
|1  |abc |short      |
|2  |xyz |short_alias|
+---+----+-----------+
...