fix header issues when setup skip first N lines

byzer-org · Aug 10, 2022 · 56e7081 · 56e7081
1 parent 4defad3
commit 56e7081
Show file tree

Hide file tree

Showing 6 changed files with 22 additions and 35 deletions.
diff --git a/streamingpro-core/src/main/java/tech/mlsql/tool/HDFSOperatorV2.scala b/streamingpro-core/src/main/java/tech/mlsql/tool/HDFSOperatorV2.scala
@@ -266,7 +266,7 @@ object HDFSOperatorV2 {
     }
   }
 
-  def saveWithoutTopNLines(inPath: String, skipFirstNLines: Int, header: Boolean): String = {
+  def saveWithoutTopNLines(inPath: String, skipFirstNLines: Int): String = {
     val fs = FileSystem.get(hadoopConfiguration)
     val src: Path = new Path(inPath)
     var br: BufferedReader = null
@@ -275,39 +275,33 @@ object HDFSOperatorV2 {
     val pathElements = inPath.split(PathFun.pathSeparator)
     val writePathParts = pathElements.take(pathElements.length - 1) :+ String.format("skipFirstNLines_%s_%s", String.valueOf(skipFirstNLines), pathElements(pathElements.length - 1))
     val outPath = writePathParts.mkString(PathFun.pathSeparator)
-    if (!fileExists(outPath)) {
-      try {
-        dos = fs.create(new Path(new java.io.File(outPath).getPath), true)
-        br = new BufferedReader(new InputStreamReader(fs.open(src)))
-        line = br.readLine()
-
-        var count = 1
-        while (line != null) {
 
-          if (header && count == 1) {
-            dos.writeBytes(line + "\n")
-            line = br.readLine()
-          }
+    try {
+      dos = fs.create(new Path(new java.io.File(outPath).getPath), true)
+      br = new BufferedReader(new InputStreamReader(fs.open(src)))
+      line = br.readLine()
 
-          if (count >= skipFirstNLines) {
-            dos.writeBytes(line + "\n")
-          }
-          count += 1
-          line = br.readLine()
+      var count = 1
+      while (line != null) {
+        if (count > skipFirstNLines) {
+          dos.writeBytes(line + "\n")
         }
-      } finally {
-        if (br != null) br.close()
-        if (null != dos) {
-          try {
-            dos.close()
-          } catch {
-            case ex: Exception =>
-              println("close exception")
-          }
+        count += 1
+        line = br.readLine()
+      }
+    } finally {
+      if (br != null) br.close()
+      if (null != dos) {
+        try {
           dos.close()
+        } catch {
+          case ex: Exception =>
+            println("close exception")
         }
+        dos.close()
       }
     }
+
     outPath
   }
 }
diff --git a/streamingpro-mlsql/libs/aws-glue-datacatalog-spark-client-3.4.0-SNAPSHOT.jar b/streamingpro-mlsql/libs/aws-glue-datacatalog-spark-client-3.4.0-SNAPSHOT.jar
diff --git a/streamingpro-mlsql/libs/hive-common-2.3.7.jar b/streamingpro-mlsql/libs/hive-common-2.3.7.jar
diff --git a/streamingpro-mlsql/libs/hive-exec-2.3.7-core.jar b/streamingpro-mlsql/libs/hive-exec-2.3.7-core.jar
diff --git a/streamingpro-mlsql/libs/hive-exec-2.3.7.jar b/streamingpro-mlsql/libs/hive-exec-2.3.7.jar
diff --git a/streamingpro-mlsql/src/main/java/streaming/core/datasource/impl/MLSQLCSV.scala b/streamingpro-mlsql/src/main/java/streaming/core/datasource/impl/MLSQLCSV.scala
@@ -20,15 +20,8 @@ class MLSQLCSV(override val uid: String) extends MLSQLBaseFileSource with WowPar
       case None => originPath
       case Some(_) => {
         var numsToSkip = skipFirstNLines.get.toInt
-        val header = config.config.get("header") match {
-          case Some("true") => {
-            numsToSkip += 1
-            true
-          }
-          case _ => false
-        }
         val path = originPath
-        val newPath = HDFSOperatorV2.saveWithoutTopNLines(path, numsToSkip, header)
+        val newPath = HDFSOperatorV2.saveWithoutTopNLines(path, numsToSkip)
         newPath
       }
     }