Spark Dataframe: Как добавить индекс Колонка: Aka Distributed Data Index

Ответ 1

С Scala вы можете использовать:

import org.apache.spark.sql.functions._ 

df.withColumn("id",monotonicallyIncreasingId)

Вы можете обратиться к этим примерам и scala docs.

С Pyspark вы можете использовать:

from pyspark.sql.functions import monotonically_increasing_id 

df_index = df.select("*").withColumn("id", monotonically_increasing_id())

Ответ 2

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

"Я хочу добавить столбец от 1 до номера строки".

Пусть говорят, что мы имеем следующий DF

+--------+-------------+-------+
| userId | productCode | count |
+--------+-------------+-------+
|     25 |        6001 |     2 |
|     11 |        5001 |     8 |
|     23 |         123 |     5 |
+--------+-------------+-------+

Чтобы сгенерировать идентификаторы, начиная с 1

val w = Window.orderBy("count")
val result = df.withColumn("index", row_number().over(w))

Это добавит индексный столбец, упорядоченный по возрастанию значения count.

+--------+-------------+-------+-------+
| userId | productCode | count | index |
+--------+-------------+-------+-------+
|     25 |        6001 |     2 |     1 |
|     23 |         123 |     5 |     2 |
|     11 |        5001 |     8 |     3 |
+--------+-------------+-------+-------+

Ответ 3

  ПРИМЕЧАНИЕ. Приведенные выше подходы не дают порядковый номер, но они увеличивают идентификатор.

Простой способ сделать это и обеспечить порядок индексов, как показано ниже. zipWithIndex.

Пример данных.

+-------------------+
|               Name|
+-------------------+
|     Ram Ghadiyaram|
|        Ravichandra|
|              ilker|
|               nick|
|             Naveed|
|      Gobinathan SP|
|Sreenivas Venigalla|
|     Jackela Kowski|
|   Arindam Sengupta|
|            Liangpi|
|             Omar14|
|        anshu kumar|
+-------------------+

    package com.example

import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{LongType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row}

/**
  * DistributedDataIndex : Program to index an RDD  with
  */
object DistributedDataIndex extends App with Logging {

  val spark = builder
    .master("local[*]")
    .appName(this.getClass.getName)
    .getOrCreate()

  import spark.implicits._

  val df = spark.sparkContext.parallelize(
    Seq("Ram Ghadiyaram", "Ravichandra", "ilker", "nick"
      , "Naveed", "Gobinathan SP", "Sreenivas Venigalla", "Jackela Kowski", "Arindam Sengupta", "Liangpi", "Omar14", "anshu kumar"
    )).toDF("Name")
  df.show
  logInfo("addColumnIndex here")
  // Add index now...
  val df1WithIndex = addColumnIndex(df)
    .withColumn("monotonically_increasing_id", monotonically_increasing_id)
  df1WithIndex.show(false)

  /**
    * Add Column Index to dataframe to each row
    */
  def addColumnIndex(df: DataFrame) = {
    spark.sqlContext.createDataFrame(
      df.rdd.zipWithIndex.map {
        case (row, index) => Row.fromSeq(row.toSeq :+ index)
      },
      // Create schema for index column
      StructType(df.schema.fields :+ StructField("index", LongType, false)))
  }
}

Результат:

+-------------------+-----+---------------------------+
|Name               |index|monotonically_increasing_id|
+-------------------+-----+---------------------------+
|Ram Ghadiyaram     |0    |0                          |
|Ravichandra        |1    |8589934592                 |
|ilker              |2    |8589934593                 |
|nick               |3    |17179869184                |
|Naveed             |4    |25769803776                |
|Gobinathan SP      |5    |25769803777                |
|Sreenivas Venigalla|6    |34359738368                |
|Jackela Kowski     |7    |42949672960                |
|Arindam Sengupta   |8    |42949672961                |
|Liangpi            |9    |51539607552                |
|Omar14             |10   |60129542144                |
|anshu kumar        |11   |60129542145                |
+-------------------+-----+---------------------------+

Ответ 4

Как сказал Рам, zippedwithindex лучше, чем монотонно увеличивающийся идентификатор, если вам нужны последовательные номера строк. Попробуйте это (среда PySpark):

from pyspark.sql import Row
from pyspark.sql.types import StructType, StructField, LongType

new_schema = StructType(**original_dataframe**.schema.fields[:] + [StructField("index", LongType(), False)])
zipped_rdd = **original_dataframe**.rdd.zipWithIndex()
indexed = (zipped_rdd.map(lambda ri: row_with_index(*list(ri[0]) + [ri[1]])).toDF(new_schema))

где original_dataframe - это фрейм данных, к которому нужно добавить индекс, а row_with_index - это новая схема с индексом столбца, который вы можете записать как

row_with_index = Row(
"calendar_date"
,"year_week_number"
,"year_period_number"
,"realization"
,"index"
)

Здесь calendar_date, year_week_number, year_period_number и реализация были столбцами моего исходного кадра данных. Вы можете заменить имена именами ваших столбцов. index - это новое имя столбца, которое вы должны были добавить для номеров строк.

Ответ 5

Как получить столбец с последовательным идентификатором id [1, 2, 3, 4... n]:

from pyspark.sql.functions import desc, row_number, monotonically_increasing_id

df_with_seq_id = df.withColumn('index_column_name', row_number().over(Window.orderBy(monotonically_increasing_id())) - 1)

Обратите внимание, что row_number() начинается с 1, поэтому вычтите на 1, если вы хотите 0-индексированный столбец