Добавить столбец в фрейм данных pyspark с неуникальными идентификаторами по другому ключу

Question

Извиняюсь за название - не знаю, как легко подвести итог моей проблемы.

У меня есть фрейм данных pyspark с 2 столбцами, кодом и emp. Каждое уникальное кодовое значение имеет несколько значений emp, как показано ниже. Я хочу добавить столбец, который для каждого уникального значения кода применяет увеличивающийся номер, например, столбец значения ниже. У меня была игра с monotonicallyIncreasingId(), и мне не удалось ограничить создание его идентификатора одним конкретным ключом кода, и действительно, в документации сказано, что индексы не должны увеличиваться по порядку.

+----+---+-----+
|code|emp|value|
+----+---+-----+
|   a| 14|    1|
|   a| 22|    2|
|   a| 35|    3|
|   a| 64|    4|
|   b| 12|    1|
...
+----+---+-----+

Максимальное значение будет равно 4 emp значениям на кодовое значение, если это будет влиять на эффективность. Индексы должны увеличиваться в зависимости от размера значения emp - самое низкое должно иметь значение 1, самое высокое значение n, где n - количество записей с конкретным кодом.

Answer 1

Вы можете создать временное представление и использовать Spark SQL для этого:

>>> df = spark.createDataFrame([('a', 14), ('a', 22), ('a', 35), ('a', 64), ('b', 12)], ['code', 'emp'])
>>> df.show()
+----+---+
|code|emp|
+----+---+
|   a| 14|
|   a| 22|
|   a| 35|
|   a| 64|
|   b| 12|
+----+---+
>>> df.createOrReplaceTempView("df")
>>> df2 = spark.sql("select code, emp, row_number() over(partition by code order by emp) as value from df order by code")
>>> df2.show()
+----+---+-----+
|code|emp|value|
+----+---+-----+                                                                                                         
|   a| 14|    1|                                                                                                             
|   a| 22|    2|                                                                                                             
|   a| 35|    3|                                                                                                          
|   a| 64|    4| 
|   b| 12|    1|                                                                                                             
+----+---+-----+

Answer 2

Вы можете использовать row_number () с Оконным управлением функциями.

Первый импорт Окно и row_number ,

from pyspark.sql import Window
from pyspark.sql.functions import row_number()

Предположим, ваш сценарий со следующими столбцами и значениями

>>> cols1 = ['code', 'emp']
>>> vals1 = [
     ('a', 14),
     ('a', 22),
     ('a', 35),
     ('a', 64),
     ('b', 12),
     ('b', 35)
]
# Create a DataFrame
>>> df1 = spark.createDataFrame(vals1, cols1)

# Result of 'df1' table.
>>> df1.show()
+----+---+
|code|emp|
+----+---+
|   a| 14|
|   a| 22|
|   a| 35|
|   a| 64|
|   b| 12|
|   b| 35|
+----+---+

Применить, row_number () над столбцом code.

>>> val = df1.withColumn("value", row_number().over(Window.partitionBy("code").orderBy("emp")))

>>> val.show()
+----+---+-----+
|code|emp|value|
+----+---+-----+
|   b| 12|    1|
|   b| 35|    2|
|   a| 14|    1|
|   a| 22|    2|
|   a| 35|    3|
|   a| 64|    4|
+----+---+-----+

Наконец, выполните заказ по столбцу code, чтобы получить желаемый результат.

>>> val.orderBy('code').show()
+----+---+-----+
|code|emp|value|
+----+---+-----+
|   a| 14|    1|
|   a| 22|    2|
|   a| 35|    3|
|   a| 64|    4|
|   b| 12|    1|
|   b| 35|    2|
+----+---+-----+

• partitionBy : создает WindowSpec с определенным разделением.

Для получения дополнительной информации см .:

• Окно : http://spark.apache.org/docs/2.2.0/api/python/pyspark.sql.html#pyspark.sql.Window
• row_number () : https://blog.jooq.org/2014/08/12/the-difference-between-row_number-rank-and-dense_rank/

Answer 3

Для Scala вы можете создать фрейм данных со столбцом инкрементного индекса, например:

%scala
val rankedWordCount = sqlContext.sql("select row_number() over (order by some_numeric_value desc) as index_col,lower(info) as info, some_numeric_value from information_table")