modification of ConvertSchema

yang qidong
2018-11-23 12:56:14 +08:00 · 2018-11-23 12:56:14 +08:00 · 2b46264473
parent cda81eae5c
commit 2b46264473
1 changed files with 2 additions and 2 deletions
--- a/piflow-bundle/src/main/scala/cn/piflow/bundle/common/ConvertSchema.scala
+++ b/piflow-bundle/src/main/scala/cn/piflow/bundle/common/ConvertSchema.scala
@ -16,14 +16,14 @@ class ConvertSchema extends ConfigurableStop {
  var schema:String = _

  def perform(in: JobInputStream, out: JobOutputStream, pec: JobContext): Unit = {
-    val df = in.read()
+    var df = in.read()

    //oldField1->newField1, oldField2->newField2
    val field = schema.split(",")

    field.foreach(f => {
      val old_new: Array[String] = f.split("->")
-      df.withColumnRenamed(old_new(0),old_new(1))
+      df = df.withColumnRenamed(old_new(0),old_new(1))
    })

    println("###########################")