La valeur de l'accumulateur d'étincelles ne s'incrémente pas

Pour cette fonctionnalité, vous pouvez utiliser numéro_ligne

import org.apache.spark.sql.expressions.Window
import static org.apache.spark.sql.functions.col;
import static org.apache.spark.sql.functions.row_number;

Dataset<Row> empDatasetWithIds = empDataset.withColumn("Ids", 
    row_number().over(Window.orderBy(col("name"), col("dept"), col("project)))
)

Référence: https://stackoverflow.com/a/31077759

Comme indiqué dans les commentaires à l'aide de Window sans partition est très inefficace. et Doit être évité dans le code de production pour traiter des données volumineuses.

Votre approche avec l'accumulateur ne fonctionne pas (comme expliqué dans Pourquoi le nœud worker ne voit-il pas les mises à jour de l'accumulateur sur un autre nœud worker? ) depuis Spark exécute ce code dans différents exécuteurs (différents processus jvm s'exécutant sur différentes machines), et chacun a sa propre copie si l'accumulateur.

Vous pouvez faire simplement comme ci-dessous, si l'aspect ascendant séquentiel n'est pas un problème:

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

val ds = sc.parallelize(Seq(
    ("a", -1.0), ("b", -2.0), ("c", -3.0)),5).toDS   // Just a dummy DS

val newds = ds.withColumn("uniqueIdColumn", monotonically_increasing_id())

newds.show(false)

Essayez-le et adaptez-vous à votre propre situation.

BTW: Wrong utilisation de l'accumulateur.

Les accumulateurs sont des variables utilisées pour accumuler des données dans les exécuteurs et les renvoyer au pilote. Si vous lisez sa valeur à partir de l'exécuteur, le comportement n'est pas défini (AFAIK). Je pense que vous obtiendriez probablement ce qui a été accumulé jusqu'à présent pour la partition locale. En effet, le but de spark est de faire des calculs parallèles. Par conséquent, lors de l'utilisation d'un accumulateur, les données sont accumulées pour chaque partition dans un accumulateur séparé qui sont ensuite fusionnées et renvoyées au pilote (paradigme de réduction de carte). Vous ne pouvez donc pas utiliser un accumulateur pour partager des informations entre les exécuteurs. Ce n'est tout simplement pas ce à quoi cela sert

Ce que vous pouvez faire, c'est utiliser zipWithIndex de l'API RDD si vous avez besoin d'index consécutifs, ou monoticallyIncreasingId de l'API SparkSQL si vous avez juste besoin d'indices croissants. Le premier déclenche un petit travail d'étincelle tandis que le second est presque gratuit (pas de travail d'étincelle).

Option 1 (augmentant mais pas nécessairement les indices consécutifs)

StructType schema = yourDataframe.schema();
schema.add(new StructField("id", DataTypes.LongType, false,null));
JavaRDD<Row> rdd = yourDataframe.toJavaRDD().zipWithIndex()
    .map(x -> {
         Collection<Object> row = JavaConverters.asJavaCollection(x._1.toSeq());
         Long index = x._2;
         row.add(index);
         return RowFactory.create(row);
    });
Dataset<Row> indexedData = spark.createDataFrame(rdd, schema);

Option 2 (indices consécutifs et croissants)

yourDataframe.withColumn("id", functions.monotonicallyIncreasingId());