Merge pull request #66 from djfreels/bug/data-frame-writer-fixes

DataFrameWriterOptions bug fix
Acxiom · Oct 4, 2019 · 3ff5ca8 · 3ff5ca8
2 parents bc06b14 + ba12166
commit 3ff5ca8
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 7 deletions.
diff --git a/common-pipeline-steps/src/main/scala/com/acxiom/pipeline/steps/DataFrameSteps.scala b/common-pipeline-steps/src/main/scala/com/acxiom/pipeline/steps/DataFrameSteps.scala
@@ -56,18 +56,18 @@ object DataFrameSteps {
       .mode(options.saveMode)
       .options(options.options.getOrElse(Map[String, String]()))
 
-    val w1 = if (options.bucketingOptions.isDefined) {
+    val w1 = if (options.bucketingOptions.isDefined && options.bucketingOptions.get.columns.nonEmpty) {
       val bucketingOptions = options.bucketingOptions.get
       writer.bucketBy(bucketingOptions.numBuckets, bucketingOptions.columns.head, bucketingOptions.columns.drop(1): _*)
     } else {
       writer
     }
-    val w2 = if (options.partitionBy.isDefined) {
+    val w2 = if (options.partitionBy.isDefined && options.partitionBy.get.nonEmpty) {
       w1.partitionBy(options.partitionBy.get: _*)
     } else {
       w1
     }
-    if (options.sortBy.isDefined) {
+    if (options.sortBy.isDefined && options.sortBy.get.nonEmpty) {
       val sortBy = options.sortBy.get
       w2.sortBy(sortBy.head, sortBy.drop(1): _*)
     } else {

diff --git a/spark-pipeline-engine/src/main/scala/com/acxiom/pipeline/PipelineStepMapper.scala b/spark-pipeline-engine/src/main/scala/com/acxiom/pipeline/PipelineStepMapper.scala
@@ -303,9 +303,9 @@ trait PipelineStepMapper {
   }
 
   private def getPathValues(value: String, pipelineContext: PipelineContext): PipelinePath = {
-    if (value.contains('.')) {
+    val special = getSpecialCharacter(value)
+    if (value.contains('.') && special.nonEmpty) {
       // Check for the special character
-      val special = getSpecialCharacter(value)
       val pipelineId = value.substring(0, value.indexOf('.')).substring(1)
       val paths = value.split('.')
       if (pipelineContext.parameters.hasPipelineParameters(pipelineId)) {

diff --git a/spark-pipeline-engine/src/test/scala/com/acxiom/pipeline/PipelineStepMapperTests.scala b/spark-pipeline-engine/src/test/scala/com/acxiom/pipeline/PipelineStepMapperTests.scala
@@ -146,19 +146,22 @@ class PipelineStepMapperTests extends FunSpec with BeforeAndAfterAll with GivenW
             Parameter(name = Some("One"), value=Some(classMap), className = Some("com.acxiom.pipeline.ParameterTest")),
             Parameter(name = Some("Two"), value=Some("15"),`type`=Some("integer")),
             Parameter(name = Some("Three")),
-            Parameter(name = Some("Four"), defaultValue = Some("Four default"))))),
+            Parameter(name = Some("Four"), defaultValue = Some("Four default")),
+            Parameter(name = Some("Five"), value=Some("silkie.chicken@chickens.com"))))),
         pipelineContext)
-      assert(parameterMap.size == 4)
+      assert(parameterMap.size == 5)
       assert(parameterMap.contains("One"))
       assert(parameterMap.contains("Two"))
       assert(parameterMap.contains("Three"))
       assert(parameterMap.contains("Four"))
+      assert(parameterMap.contains("Five"))
       assert(parameterMap("One").isInstanceOf[ParameterTest])
       assert(parameterMap("One").asInstanceOf[ParameterTest].string.getOrElse("") == "fred")
       assert(parameterMap("One").asInstanceOf[ParameterTest].num.getOrElse(0) == 3)
       assert(parameterMap("Two").asInstanceOf[Int] == 15)
       assert(parameterMap("Three").asInstanceOf[Option[_]].isEmpty)
       assert(parameterMap("Four").asInstanceOf[String] == "Four default")
+      assert(parameterMap("Five").asInstanceOf[String] == "silkie.chicken@chickens.com")
     }
 
     it("Should create object with variables") {