Suppression des doublons de lignes en fonction des colonnes spécifiques dans un EDR/Spark DataFrame

De votre question, il est difficile-pour les colonnes que vous souhaitez utiliser pour déterminer les doublons. L'idée générale derrière la solution est de créer une clé basée sur les valeurs des colonnes qui permettent d'identifier les doublons. Ensuite, vous pouvez utiliser le reduceByKey ou de réduire les opérations d'éliminer les doublons.

Voici un peu de code pour vous aider à démarrer:

def get_key(x):
    return "{0}{1}{2}".format(x[0],x[2],x[3])

m = data.map(lambda x: (get_key(x),x))

Maintenant, vous avez une clé-valeur RDD qui sont saisies par les colonnes 1, 3 et 4.
La prochaine étape serait un reduceByKey ou groupByKey et filter.
Cela permettrait d'éliminer les doublons.

r = m.reduceByKey(lambda x,y: (x))

D'accord avec David. Pour ajouter, il peut pas être le cas que nous voulons groupBy toutes les colonnes autres que la colonne(s) en fonction d'agrégation j'.e, si nous voulons supprimer les doublons purement basée sur un sous-ensemble de colonnes et de conserver toutes les colonnes dans l'original dataframe. Donc, la meilleure façon de le faire pourrait être à l'aide de dropDuplicates Dataframe api disponible dans Spark 1.4.0

Pour référence, voir: https://spark.apache.org/docs/1.4.0/api/scala/index.html#org.apache.spark.sql.DataFrame

Le programme ci-dessous vous aidera à déposer en double dans son ensemble , ou si vous voulez supprimer les doublons en fonction de certaines colonnes , vous pouvez même le faire:

import org.apache.spark.sql.SparkSession

object DropDuplicates {
def main(args: Array[String]) {
val spark =
  SparkSession.builder()
    .appName("DataFrame-DropDuplicates")
    .master("local[4]")
    .getOrCreate()

import spark.implicits._

//create an RDD of tuples with some data
val custs = Seq(
  (1, "Widget Co", 120000.00, 0.00, "AZ"),
  (2, "Acme Widgets", 410500.00, 500.00, "CA"),
  (3, "Widgetry", 410500.00, 200.00, "CA"),
  (4, "Widgets R Us", 410500.00, 0.0, "CA"),
  (3, "Widgetry", 410500.00, 200.00, "CA"),
  (5, "Ye Olde Widgete", 500.00, 0.0, "MA"),
  (6, "Widget Co", 12000.00, 10.00, "AZ")
)
val customerRows = spark.sparkContext.parallelize(custs, 4)

//convert RDD of tuples to DataFrame by supplying column names
val customerDF = customerRows.toDF("id", "name", "sales", "discount", "state")

println("*** Here's the whole DataFrame with duplicates")

customerDF.printSchema()

customerDF.show()

//drop fully identical rows
val withoutDuplicates = customerDF.dropDuplicates()

println("*** Now without duplicates")

withoutDuplicates.show()

//drop fully identical rows
val withoutPartials = customerDF.dropDuplicates(Seq("name", "state"))

println("*** Now without partial duplicates too")

withoutPartials.show()

 }
 }