Comment puis-je ajouter un persistante de la colonne d'id de ligne à Étincelle DataFrame?

Spark 2.0

• Ce problème a été résolu dans Spark 2.0 avec SPARK-14241.

• Une autre question similaire a été résolu dans Spark 2.1 avec SPARK-14393

Étincelle 1.x

Problème que vous rencontrez est assez subtil, mais peut être réduite à un simple fait monotonically_increasing_id est extrêmement laide de la fonction. Il n'est clairement pas pur, et sa valeur dépend de quelque chose qui est complètement hors de votre contrôle.

Il ne prend pas tous les paramètres afin de un optimiseur de point de vue, il n'a pas d'importance quand il est appelé, et peut être poussé après tous les autres opérations. D'où le comportement que vous voyez.

Si vous prenez regarder le code que vous trouverez ce qui est explicitement marquée par l'extension de MonotonicallyIncreasingID expression avec Nondeterministic.

Je ne pense pas qu'il existe une solution élégante mais d'une façon que vous pouvez gérer cela est d'ajouter une dépendance artificielle sur la valeur filtrée. Par exemple avec un UDF comme ceci:

from pyspark.sql.types import LongType
from pyspark.sql.functions import udf

bound = udf(lambda _, v: v, LongType()) 

(df
  .withColumn("rn", monotonically_increasing_id())
  # Due to nondeterministic behavior it has to be a separate step
  .withColumn("rn", bound("P", "rn"))  
  .where("P"))

En général, il pourrait être plus propre pour ajouter des indices à l'aide de zipWithIndex sur un RDD puis convertir un DataFrame.

* Solution de contournement ci-dessus n'est plus une solution valable (ni) dans Spark 2.x où Python Udf sont l'objet de l'exécution du plan d'optimisations.

J'ai été récemment travaillé sur un problème similaire. Bien que monotonically_increasing_id() est très rapide, il n'est pas fiable et ne va pas vous donner consécutive, les numéros de ligne, seulement une augmentation unique des entiers.

La création d'une partition windows et ensuite à l'aide de row_number().over(some_windows_partition) est très coûteux en temps.

La meilleure solution pour l'instant est à l'aide de zippée avec l'index, puis convertir le fichier zippé en arrière à l'original dataframe, avec le nouveau schéma, y compris l'indice de colonne.

Essayez ceci:

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))

Où original_dataframe est le dataframe vous devez ajouter un index sur et row_with_index est le nouveau schéma avec l'index de colonne que vous pouvez écrire comme

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

Ici, calendar_date, year_week_number, year_period_number, et realization ont été les colonnes de mon origine dataframe. Vous pouvez remplacer les noms avec les noms de vos colonnes. L'indice est le nouveau nom de la colonne que vous deviez ajouter pour les numéros de ligne.

De ce processus est largement plus efficace et plus facile en comparaison avec row_number().over(some_windows_partition) méthode.

Espère que cette aide.

Cela a fonctionné pour moi. Créé une autre colonne d'identité et utilisé la fenêtre de la fonction row_number

import org.apache.spark.sql.functions.{row_number}
import org.apache.spark.sql.expressions.Window

val df1: DataFrame = df.withColumn("Id",lit(1))

df1
.select(
...,
row_number()
.over(Window
.partitionBy("Id"
.orderBy(col("...").desc))
)
.alias("Row_Nbr")
)