Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-7324][SQL] Add DataFrame.dropDuplicates #5870

Closed
wants to merge 11 commits into from
Closed

[SPARK-7324][SQL] Add DataFrame.dropDuplicates #5870

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
d11d5b9
[SPARK-7294] ADD BETWEEN
May 1, 2015
baf839b
[SPARK-7294] ADD BETWEEN
May 2, 2015
7d62368
[SPARK-7294] ADD BETWEEN
May 2, 2015
76f0c51
Merge remote-tracking branch 'remotes/upstream/master'
May 2, 2015
f080f8d
update pep8
May 2, 2015
c6e49bc
undo
May 3, 2015
d6cc28d
undo
May 3, 2015
aab51ef
update
May 3, 2015
b6f1879
update
May 3, 2015
571869e
Remove useless code.
kaka1992 May 6, 2015
1de8791
Update
kaka1992 May 12, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/DataFrame.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -887,6 +887,22 @@ class DataFrame private[sql](
}
}

/**
* Returns a new [[DataFrame]] without duplicates.
* @group dfops
*/
def dropDuplicates(): DataFrame = dropDuplicates(this.columns)

/**
* Returns a new [[DataFrame]] without duplicates under the given columns.
* @group dfops
*/
def dropDuplicates(subset: Seq[String]): DataFrame = {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suppose that distinct is as same as dropDuplicates for removing duplicate rows? If they are the same, which implementation is better? GroupedData or Distinct node?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@viirya No, dropDuplicates is used to remove duplicate rows that are the same in some columns or in all columns (default) . The default version is as same as distinct.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You can also select subset of columns and then do distinct?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you do this, you can't get all columns.

import org.apache.spark.sql.functions.{first => columnFirst}
val columnFirsts = columns.map(columnFirst)
groupBy(subset.head, subset.tail : _*).agg(columnFirsts.head, columnFirsts.tail : _*)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we should also check if subset.size == 0 or columns.size == 0, then simply return an empty data frame (there is one in SQLContext).

}

/**
* Computes statistics for numeric columns, including count, mean, stddev, min, and max.
* If no columns are given, this function computes statistics for all numerical columns.
Expand Down
35 changes: 35 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -613,4 +613,39 @@ class DataFrameSuite extends QueryTest {
Row(new java.math.BigDecimal(2.0)))
TestSQLContext.setConf(SQLConf.CODEGEN_ENABLED, originalValue.toString)
}

test("SPARK-7324 dropDuplicates") {
val testData = TestSQLContext.sparkContext.parallelize(
(2, 1, 2) :: (1, 1, 1) ::
(1, 2, 1) :: (2, 1, 2) ::
(2, 2, 2) :: (2, 2, 1) ::
(2, 1, 1) :: (1, 1, 2) ::
(1, 2, 2) :: (1, 2, 1) :: Nil).toDF("key", "value1", "value2")

checkAnswer(
testData.dropDuplicates(),
Seq(Row(2, 1, 2), Row(1, 1, 1), Row(1, 2, 1),
Row(2, 2, 2), Row(2, 1, 1), Row(2, 2, 1),
Row(1, 1, 2), Row(1, 2, 2)))

checkAnswer(
testData.dropDuplicates(Seq("key", "value1")),
Seq(Row(2, 1, 2), Row(1, 2, 1), Row(1, 1, 1), Row(2, 2, 2)))

checkAnswer(
testData.dropDuplicates(Seq("value1", "value2")),
Seq(Row(2, 1, 2), Row(1, 2, 1), Row(1, 1, 1), Row(2, 2, 2)))

checkAnswer(
testData.dropDuplicates(Seq("key")),
Seq(Row(2, 1, 2), Row(1, 1, 1)))

checkAnswer(
testData.dropDuplicates(Seq("value1")),
Seq(Row(2, 1, 2), Row(1, 2, 1)))

checkAnswer(
testData.dropDuplicates(Seq("value2")),
Seq(Row(2, 1, 2), Row(1, 1, 1)))
}
}