You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafu.apache.org by ey...@apache.org on 2018/11/15 14:47:29 UTC
datafu git commit: Add tests for flatten and changeSchema
Repository: datafu
Updated Branches:
refs/heads/spark-tmp 65b49fade -> 986af4540
Add tests for flatten and changeSchema
Project: http://git-wip-us.apache.org/repos/asf/datafu/repo
Commit: http://git-wip-us.apache.org/repos/asf/datafu/commit/986af454
Tree: http://git-wip-us.apache.org/repos/asf/datafu/tree/986af454
Diff: http://git-wip-us.apache.org/repos/asf/datafu/diff/986af454
Branch: refs/heads/spark-tmp
Commit: 986af45404c83f35fbd2d17c490d77c905a71bc3
Parents: 65b49fa
Author: Eyal Allweil <ey...@apache.org>
Authored: Thu Nov 15 16:46:57 2018 +0200
Committer: Eyal Allweil <ey...@apache.org>
Committed: Thu Nov 15 16:46:57 2018 +0200
----------------------------------------------------------------------
.../main/scala/datafu/spark/DataFrameOps.scala | 2 --
.../main/scala/datafu/spark/SparkDFUtils.scala | 7 +++++
.../scala/datafu/spark/TestSparkDFUtils.scala | 30 ++++++++++++++++++--
3 files changed, 34 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/datafu/blob/986af454/datafu-spark/src/main/scala/datafu/spark/DataFrameOps.scala
----------------------------------------------------------------------
diff --git a/datafu-spark/src/main/scala/datafu/spark/DataFrameOps.scala b/datafu-spark/src/main/scala/datafu/spark/DataFrameOps.scala
index ddc3da6..6f5ae18 100644
--- a/datafu-spark/src/main/scala/datafu/spark/DataFrameOps.scala
+++ b/datafu-spark/src/main/scala/datafu/spark/DataFrameOps.scala
@@ -37,8 +37,6 @@ object DataFrameOps {
def changeSchema(newScheme: String*): DataFrame = SparkDFUtils.changeSchema(df, newScheme: _*)
- def filterOut(str: String): DataFrame = df.filter("not (" + str + ")")
-
def joinWithRange(colSingle: String, dfRange: DataFrame, colRangeStart: String, colRangeEnd: String, DECREASE_FACTOR: Long = 2^8) =
SparkDFUtils.joinWithRange(df, colSingle, dfRange, colRangeStart, colRangeEnd, DECREASE_FACTOR)
http://git-wip-us.apache.org/repos/asf/datafu/blob/986af454/datafu-spark/src/main/scala/datafu/spark/SparkDFUtils.scala
----------------------------------------------------------------------
diff --git a/datafu-spark/src/main/scala/datafu/spark/SparkDFUtils.scala b/datafu-spark/src/main/scala/datafu/spark/SparkDFUtils.scala
index 6d79da3..a71ec32 100644
--- a/datafu-spark/src/main/scala/datafu/spark/SparkDFUtils.scala
+++ b/datafu-spark/src/main/scala/datafu/spark/SparkDFUtils.scala
@@ -93,6 +93,13 @@ object SparkDFUtils extends SparkDFUtilsTrait {
df2
}
+ /**
+ * Returns a DataFrame with the given column (should be a StructType) replaced by its inner fields.
+ * This method only flattens a single level of nesting.
+ * @param df DataFrame to operate on
+ * @param colName column name for a column of type StructType
+ * @return DataFrame representing the data after the operation
+ */
override def flatten(df: DataFrame, colName: String): DataFrame = {
val outerFields = df.schema.fields.map(_.name).toSet
val flattenFields = df.schema(colName).dataType.asInstanceOf[StructType].fields.filter(f => !outerFields.contains(f.name)).map("`" + colName + "`.`" + _.name + "`")
http://git-wip-us.apache.org/repos/asf/datafu/blob/986af454/datafu-spark/src/test/scala/datafu/spark/TestSparkDFUtils.scala
----------------------------------------------------------------------
diff --git a/datafu-spark/src/test/scala/datafu/spark/TestSparkDFUtils.scala b/datafu-spark/src/test/scala/datafu/spark/TestSparkDFUtils.scala
index 961c6d5..37a2270 100644
--- a/datafu-spark/src/test/scala/datafu/spark/TestSparkDFUtils.scala
+++ b/datafu-spark/src/test/scala/datafu/spark/TestSparkDFUtils.scala
@@ -35,7 +35,7 @@ import scala.collection.mutable.WrappedArray
import org.apache.spark.sql.types._
@RunWith(classOf[JUnitRunner])
-class DedupTests extends FunSuite with DataFrameSuiteBase {
+class DataFrameOpsTests extends FunSuite with DataFrameSuiteBase {
import DataFrameOps._
@@ -51,8 +51,10 @@ class DedupTests extends FunSuite with DataFrameSuiteBase {
//var df : DataFrame = spark.read.format("csv").option("header", "true").load("src/test/resources/dedup.csv")
- lazy val inputDataFrame = sc.parallelize(List(("a", 1, "asd1"), ("a", 2, "asd2"), ("a", 3, "asd3"), ("b", 1, "asd4"))).toDF("col_grp", "col_ord", "col_str").cache
-
+ lazy val inputRDD = sc.parallelize(List(("a", 1, "asd1"), ("a", 2, "asd2"), ("a", 3, "asd3"), ("b", 1, "asd4")))
+
+ lazy val inputDataFrame = inputRDD.toDF("col_grp", "col_ord", "col_str").cache
+
test("dedup") {
val expected : DataFrame = sc.parallelize(List(("b",1),("a",3))).toDF("col_grp", "col_ord")
@@ -201,6 +203,28 @@ class DedupTests extends FunSuite with DataFrameSuiteBase {
assertDataFrameEquals(expected2, actual2)
}
+ test("test_changeSchema") {
+
+ val actual = inputDataFrame.changeSchema("fld1","fld2", "fld3")
+
+ val expected : DataFrame = inputRDD.toDF("fld1","fld2", "fld3")
+
+ actual.show
+
+ assertDataFrameEquals(expected, actual)
+ }
+
+ test("test_flatten") {
+
+ val input = inputDataFrame.withColumn("struct_col", expr("struct(col_grp, col_ord)")).select("struct_col")
+
+ val expected : DataFrame = inputDataFrame.select("col_grp", "col_ord")
+
+ val actual = input.flatten("struct_col")
+
+ assertDataFrameEquals(expected, actual)
+ }
+
def compare(expected: DataFrame, actual: DataFrame) = {
println ("expected: " + expected.schema)