Eliminar duplicados de filas basados en columnas específicas en un DataFrame RDD/Spark

De su pregunta, no está claro cómo-a qué columnas desea utilizar para determinar los duplicados. La idea general detrás de la solución es crear una clave basada en los valores de las columnas que identifican duplicados. Luego, puede usar las operaciones reduceByKey o reducir para eliminar duplicados.

Aquí hay un código para comenzar:

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

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

Ahora, tiene un valor clave RDD que está marcado por las columnas 1, 3 y 4. El siguiente paso sería un reduceByKey o groupByKey y filter. Este eliminaría duplicados.

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

De acuerdo con David. Para agregar, puede que no sea el caso de que queramos agrupar todas las columnas que no sean las columnas en la función aggregate, es decir, si queremos eliminar duplicados puramente basados en un subconjunto de columnas y conservar todas las columnas en el dataframe original. Así que la mejor manera de hacer esto podría ser usando dropDuplicates Dataframe api disponible en Spark 1.4.0

Para referencia, ver: https://spark.apache.org/docs/1.4.0/api/scala/index.html#org.apache.spark.sql.DataFrame

Sé que ya aceptaste la otra respuesta, pero si quieres hacer esto como un DataFrame, solo usa groupBy y agg. Suponiendo que ya tiene un DF creado (con columnas llamadas "col1", "col2", etc.) podría hacer:

myDF.groupBy($"col1", $"col3", $"col4").agg($"col1", max($"col2"), $"col3", $"col4")

Tenga en cuenta que en este caso, elegí el máximo de col2, pero podría hacer avg, min, etc.

Usé la función incorporada dropDuplicates(). Código Scala que figura a continuación

val data = sc.parallelize(List(("Foo",41,"US",3),
("Foo",39,"UK",1),
("Bar",57,"CA",2),
("Bar",72,"CA",2),
("Baz",22,"US",6),
("Baz",36,"US",6))).toDF("x","y","z","count")

data.dropDuplicates(Array("x","count")).show()

Salida:

+---+---+---+-----+
|  x|  y|  z|count|
+---+---+---+-----+
|Baz| 22| US|    6|
|Foo| 39| UK|    1|
|Foo| 41| US|    3|
|Bar| 57| CA|    2|
+---+---+---+-----+

Este es mi Df contiene 4 se repite dos veces por lo que aquí se eliminarán los valores repetidos.

scala> df.show
+-----+
|value|
+-----+
|    1|
|    4|
|    3|
|    5|
|    4|
|   18|
+-----+

scala> val newdf=df.dropDuplicates

scala> newdf.show
+-----+
|value|
+-----+
|    1|
|    3|
|    5|
|    4|
|   18|
+-----+

El siguiente programa le ayudará a soltar duplicados en su totalidad, o si desea soltar duplicados basados en ciertas columnas , incluso puede hacerlo:

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

 }
 }