This is an automated email from the ASF dual-hosted git repository. aiceflower pushed a commit to branch release-0.9.4 in repository https://gitbox.apache.org/repos/asf/linkis.git
commit c6eaa1c4226477e3486557e1d93915aeed91d848 Author: jftang <[email protected]> AuthorDate: Wed Jun 3 18:35:04 2020 +0800 storage add sub module 'source' --- .../linkis/storage/csv/CSVFsWriter.scala | 9 +- .../linkis/storage/csv/StorageCSVWriter.scala | 77 +++++------- .../linkis/storage/excel/ExcelFsWriter.scala | 17 ++- .../linkis/storage/excel/StorageExcelWriter.scala | 53 +++++--- .../linkis/storage/script/ScriptFsWriter.scala | 18 +-- .../script/compaction/PYScriptCompaction.scala | 2 +- .../script/reader/StorageScriptFsReader.scala | 36 +++--- .../script/writer/StorageScriptFsWriter.scala | 40 ++++-- .../linkis/storage/source/AbstractFileSource.scala | 136 +++++++++++++++++++++ .../linkis/storage/source/FileSource.scala | 86 +++++++++++++ .../ResultsetFileSource.scala} | 29 ++--- .../linkis/storage/source/TextFileSource.scala | 52 ++++++++ .../storage/utils/StorageConfiguration.scala | 6 +- .../linkis/storage/utils/StorageUtils.scala | 10 +- 14 files changed, 428 insertions(+), 143 deletions(-) diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/csv/CSVFsWriter.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/csv/CSVFsWriter.scala index 66dc4b123d..15fdef9829 100644 --- a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/csv/CSVFsWriter.scala +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/csv/CSVFsWriter.scala @@ -16,7 +16,7 @@ package com.webank.wedatasphere.linkis.storage.csv -import java.io.InputStream +import java.io.OutputStream import com.webank.wedatasphere.linkis.common.io.FsWriter @@ -26,14 +26,9 @@ import com.webank.wedatasphere.linkis.common.io.FsWriter abstract class CSVFsWriter extends FsWriter { val charset: String val separator: String - var isLastRow: Boolean = false - - def setIsLastRow(value: Boolean): Unit - - def getCSVStream: InputStream } object CSVFsWriter { - def getCSVFSWriter(charset: String, separator: String): CSVFsWriter = new StorageCSVWriter(charset, separator) + def getCSVFSWriter(charset: String, separator: String, outputStream: OutputStream): CSVFsWriter = new StorageCSVWriter(charset, separator, outputStream) } diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/csv/StorageCSVWriter.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/csv/StorageCSVWriter.scala index d903c10e34..5a08674d5e 100644 --- a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/csv/StorageCSVWriter.scala +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/csv/StorageCSVWriter.scala @@ -17,79 +17,60 @@ package com.webank.wedatasphere.linkis.storage.csv import java.io._ -import java.util -import java.util.Collections import com.webank.wedatasphere.linkis.common.io.{MetaData, Record} import com.webank.wedatasphere.linkis.common.utils.Logging -import com.webank.wedatasphere.linkis.storage.domain.DataType import com.webank.wedatasphere.linkis.storage.resultset.table.{TableMetaData, TableRecord} +import org.apache.commons.io.IOUtils /** * Created by johnnwang on 2018/11/12. */ -class StorageCSVWriter(charsetP: String, separatorP: String) extends CSVFsWriter with Logging { - - override val charset: String = charsetP - override val separator: String = separatorP +class StorageCSVWriter(val charset: String, val separator: String, val outputStream: OutputStream) extends CSVFsWriter with Logging { private val delimiter = separator match { case "," => ',' - case _ =>'\t' + case _ => '\t' } - private val cSVWriter = new util.ArrayList[InputStream]() - private var stream: SequenceInputStream = _ - private val buffer: StringBuilder = new StringBuilder(40000) - private var counter: Int = _ - - override def setIsLastRow(value: Boolean): Unit = {} - - def collectionInputStream(content: Array[String]): Unit = { - content.foreach(f => counter += f.length) - counter += content.size - if (counter >= 40000) { - cSVWriter.add(new ByteArrayInputStream(buffer.toString().getBytes(charset))) - buffer.clear() - content.indices.map { - case 0 => content(0) - case i => delimiter + content(i) - }.foreach(buffer.append) - buffer.append("\n") - counter = buffer.length - } else { - content.indices.map { - case 0 => content(0) - case i => delimiter + content(i) - }.foreach(buffer.append) - buffer.append("\n") - counter = buffer.length - } - } + private val buffer: StringBuilder = new StringBuilder(50000) @scala.throws[IOException] override def addMetaData(metaData: MetaData): Unit = { val head = metaData.asInstanceOf[TableMetaData].columns.map(_.columnName) - collectionInputStream(head) + write(head) + //IOUtils.write(compact(head).getBytes(charset),outputStream) + } + + private def compact(row: Array[String]): String = { + val tmp = row.foldLeft("")((l, r) => l + delimiter + r) + tmp.substring(1, tmp.length) + "\n" + } + + private def write(row: Array[String]) = { + val cotent: String = compact(row) + if (buffer.length + cotent.length > 49500) { + IOUtils.write(buffer.toString().getBytes(charset), outputStream) + buffer.clear() + } + buffer.append(cotent) } @scala.throws[IOException] override def addRecord(record: Record): Unit = { - val body = record.asInstanceOf[TableRecord].row.map(f => if (f != null) f.toString else DataType.NULL_VALUE) - collectionInputStream(body) + val body = record.asInstanceOf[TableRecord].row.map(_.toString) //read时候进行null替换等等 + write(body) + //IOUtils.write(compact(body).getBytes(charset),outputStream) } - override def flush(): Unit = {} + override def flush(): Unit = { + IOUtils.write(buffer.toString().getBytes(charset), outputStream) + buffer.clear() + } override def close(): Unit = { - if (stream != null) stream.close() + flush() + IOUtils.closeQuietly(outputStream) } - override def getCSVStream: InputStream = { - cSVWriter.add(new ByteArrayInputStream(buffer.toString().getBytes(charset))) - buffer.clear() - counter = 0 //Subsequent move to flush(后续挪到flush) - stream = new SequenceInputStream(Collections.enumeration(cSVWriter)) - stream - } } diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/excel/ExcelFsWriter.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/excel/ExcelFsWriter.scala index 6c6afbb899..5901440272 100644 --- a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/excel/ExcelFsWriter.scala +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/excel/ExcelFsWriter.scala @@ -16,21 +16,20 @@ package com.webank.wedatasphere.linkis.storage.excel +import java.io.OutputStream + import com.webank.wedatasphere.linkis.common.io.FsWriter -import org.apache.poi.ss.usermodel.Workbook /** * Created by johnnwang on 2018/11/12. */ -abstract class ExcelFsWriter extends FsWriter{ - val charset:String - val sheetName:String - val dateFormat:String - - def getWorkBook: Workbook +abstract class ExcelFsWriter extends FsWriter { + val charset: String + val sheetName: String + val dateFormat: String } -object ExcelFsWriter{ - def getExcelFsWriter(charset:String,sheetName:String,dateFormat:String):ExcelFsWriter = new StorageExcelWriter(charset,sheetName,dateFormat) +object ExcelFsWriter { + def getExcelFsWriter(charset: String, sheetName: String, dateFormat: String, outputStream: OutputStream): ExcelFsWriter = new StorageExcelWriter(charset, sheetName, dateFormat, outputStream: OutputStream) } diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/excel/StorageExcelWriter.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/excel/StorageExcelWriter.scala index 69c9d8e622..6b3ba726d8 100644 --- a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/excel/StorageExcelWriter.scala +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/excel/StorageExcelWriter.scala @@ -16,14 +16,13 @@ package com.webank.wedatasphere.linkis.storage.excel -import java.io.IOException -import java.text.SimpleDateFormat +import java.io._ import java.util import com.webank.wedatasphere.linkis.common.io.{MetaData, Record} -import com.webank.wedatasphere.linkis.common.utils.Utils import com.webank.wedatasphere.linkis.storage.domain.DataType import com.webank.wedatasphere.linkis.storage.resultset.table.{TableMetaData, TableRecord} +import org.apache.commons.io.IOUtils import org.apache.poi.ss.usermodel._ import org.apache.poi.xssf.streaming.{SXSSFSheet, SXSSFWorkbook} @@ -32,10 +31,7 @@ import scala.collection.mutable.ArrayBuffer /** * Created by johnnwang on 2018/11/12. */ -class StorageExcelWriter(charsetP: String, sheetNameP: String, dateFormatP: String) extends ExcelFsWriter { - override val charset: String = charsetP - override val sheetName: String = sheetNameP - override val dateFormat: String = dateFormatP +class StorageExcelWriter(val charset: String, val sheetName: String, val dateFormat: String, val outputStream: OutputStream) extends ExcelFsWriter { private var workBook: SXSSFWorkbook = _ private var sheet: SXSSFSheet = _ @@ -43,7 +39,10 @@ class StorageExcelWriter(charsetP: String, sheetNameP: String, dateFormatP: Stri private var types: Array[DataType] = _ private var rowPoint = 0 private var columnCounter = 0 - private val styles = new util.HashMap[String,CellStyle]() + private val styles = new util.HashMap[String, CellStyle]() + private var isFlush = true + private val os = new ByteArrayOutputStream() + private var is: ByteArrayInputStream = _ def init = { workBook = new SXSSFWorkbook() @@ -61,8 +60,8 @@ class StorageExcelWriter(charsetP: String, sheetNameP: String, dateFormatP: Stri } def getWorkBook: Workbook = { - //Adaptive column width(自适应列宽) - sheet.trackAllColumnsForAutoSizing(); + //自适应列宽 + sheet.trackAllColumnsForAutoSizing() for (elem <- 0 to columnCounter) { sheet.autoSizeColumn(elem) } @@ -78,13 +77,13 @@ class StorageExcelWriter(charsetP: String, sheetNameP: String, dateFormatP: Stri style } - def getCellStyle(dataType: DataType):CellStyle = { + def getCellStyle(dataType: DataType): CellStyle = { val style = styles.get(dataType.typeName) if (style == null) { val newStyle = createCellStyle(dataType) - styles.put(dataType.typeName,newStyle) + styles.put(dataType.typeName, newStyle) newStyle - } else{ + } else { style } } @@ -109,16 +108,14 @@ class StorageExcelWriter(charsetP: String, sheetNameP: String, dateFormatP: Stri @scala.throws[IOException] override def addRecord(record: Record): Unit = { - // TODO: Do you need to replace the null value?(是否需要替换null值) + // TODO: 是否需要替换null值 val tableBody = sheet.createRow(rowPoint) var colunmPoint = 0 val excelRecord = record.asInstanceOf[TableRecord].row for (elem <- excelRecord) { val cell = tableBody.createCell(colunmPoint) val dataType = types.apply(colunmPoint) - dataType.toString match { - case _ =>if(elem !=null) cell.setCellValue(elem.toString) else cell.setCellValue(DataType.NULL_VALUE) - } + cell.setCellValue(elem.toString) //read时候进行null替换等等 cell.setCellStyle(getCellStyle(dataType)) colunmPoint += 1 } @@ -126,10 +123,28 @@ class StorageExcelWriter(charsetP: String, sheetNameP: String, dateFormatP: Stri } - override def flush(): Unit = {} + override def flush(): Unit = { + getWorkBook.write(os) + val content: Array[Byte] = os.toByteArray + is = new ByteArrayInputStream(content) + val buffer: Array[Byte] = new Array[Byte](1024) + var bytesRead: Int = 0 + while (isFlush) { + bytesRead = is.read(buffer, 0, 1024) + if (bytesRead == -1) { + isFlush = false + } else { + outputStream.write(buffer, 0, bytesRead) + } + } + } override def close(): Unit = { - Utils.tryFinally(if (workBook != null) workBook.close())() + if (isFlush) flush() + IOUtils.closeQuietly(outputStream) + IOUtils.closeQuietly(is) + IOUtils.closeQuietly(os) + IOUtils.closeQuietly(workBook) } } diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/ScriptFsWriter.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/ScriptFsWriter.scala index 4df2ac5277..43b3a326a5 100644 --- a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/ScriptFsWriter.scala +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/ScriptFsWriter.scala @@ -16,7 +16,7 @@ package com.webank.wedatasphere.linkis.storage.script -import java.io.OutputStream +import java.io.{InputStream, OutputStream} import com.webank.wedatasphere.linkis.common.io.{FsPath, FsWriter, MetaData} import com.webank.wedatasphere.linkis.storage.LineRecord @@ -28,25 +28,27 @@ abstract class ScriptFsWriter extends FsWriter { val path: FsPath val charset: String - val outputStream: OutputStream + + def getInputStream(): InputStream } object ScriptFsWriter { - def getScriptFsWriter(path: FsPath, charset: String,outputStream: OutputStream): ScriptFsWriter = new StorageScriptFsWriter(path, charset,outputStream) + def getScriptFsWriter(path: FsPath, charset: String, outputStream: OutputStream = null): ScriptFsWriter = + new StorageScriptFsWriter(path, charset, outputStream) } object ParserFactory { - def listParsers(): Array[Parser] = Array(PYScriptParser(), QLScriptParser(),ScalaScriptParser()) + def listParsers(): Array[Parser] = Array(PYScriptParser(), QLScriptParser(), ScalaScriptParser()) } object Compaction { - def listCompactions(): Array[Compaction] = Array(PYScriptCompaction(),QLScriptCompaction(),ScalaScriptCompaction()) + def listCompactions(): Array[Compaction] = Array(PYScriptCompaction(), QLScriptCompaction(), ScalaScriptCompaction()) } trait Parser { - def prefixConf:String + def prefixConf: String def prefix: String @@ -57,7 +59,7 @@ trait Parser { trait Compaction { - def prefixConf:String + def prefixConf: String def prefix: String @@ -79,5 +81,5 @@ class ScriptMetaData(var variables: Array[Variable]) extends MetaData { class ScriptRecord(line: String) extends LineRecord(line) //definition variable; specialConfiguration ;runConfiguration; startUpConfiguration; -case class Variable(sortParent:String,sort:String,key: String, value: String) +case class Variable(sortParent: String, sort: String, key: String, value: String) diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/compaction/PYScriptCompaction.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/compaction/PYScriptCompaction.scala index 9803d9c6a0..0b2f4a134c 100644 --- a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/compaction/PYScriptCompaction.scala +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/compaction/PYScriptCompaction.scala @@ -23,7 +23,7 @@ class PYScriptCompaction private extends CommonScriptCompaction { override def belongTo(suffix: String): Boolean = { suffix match { - case "python"|"py" => true + case "python"|"py"|"sh" => true case _ => false } } diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/reader/StorageScriptFsReader.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/reader/StorageScriptFsReader.scala index b3e826c9fc..f12675340c 100644 --- a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/reader/StorageScriptFsReader.scala +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/reader/StorageScriptFsReader.scala @@ -21,6 +21,7 @@ import java.io._ import com.webank.wedatasphere.linkis.common.io.{FsPath, MetaData, Record} import com.webank.wedatasphere.linkis.storage.script._ import com.webank.wedatasphere.linkis.storage.utils.StorageUtils +import org.apache.commons.io.IOUtils import scala.collection.mutable.ArrayBuffer @@ -28,14 +29,10 @@ import scala.collection.mutable.ArrayBuffer /** * Created by johnnwang on 2018/10/23. */ -class StorageScriptFsReader(pathP: FsPath, charsetP: String, inputStreamP: InputStream) extends ScriptFsReader { +class StorageScriptFsReader(val path: FsPath, val charset: String, val inputStream: InputStream) extends ScriptFsReader { - override val path: FsPath = pathP - override val charset: String = charsetP - - private val inputStream: InputStream = inputStreamP private var inputStreamReader: InputStreamReader = _ - private var BufferedReader: BufferedReader = _ + private var bufferedReader: BufferedReader = _ private var metadata: ScriptMetaData = _ @@ -47,7 +44,7 @@ class StorageScriptFsReader(pathP: FsPath, charsetP: String, inputStreamP: Input if (metadata == null) throw new IOException("Must read metadata first(必须先读取metadata)") val record = new ScriptRecord(lineText) - lineText = BufferedReader.readLine() + lineText = bufferedReader.readLine() record } @@ -55,10 +52,10 @@ class StorageScriptFsReader(pathP: FsPath, charsetP: String, inputStreamP: Input override def getMetaData: MetaData = { if (metadata == null) init() val parser = ParserFactory.listParsers().filter(p => p.belongTo(StorageUtils.pathToSuffix(path.getPath))) - lineText = BufferedReader.readLine() - while (hasNext && parser.length > 0 && isMetadata(lineText, parser(0).prefix,parser(0).prefixConf)) { + lineText = bufferedReader.readLine() + while (hasNext && parser.length > 0 && isMetadata(lineText, parser(0).prefix, parser(0).prefixConf)) { variables += parser(0).parse(lineText) - lineText = BufferedReader.readLine() + lineText = bufferedReader.readLine() } metadata = new ScriptMetaData(variables.toArray) metadata @@ -66,7 +63,7 @@ class StorageScriptFsReader(pathP: FsPath, charsetP: String, inputStreamP: Input def init(): Unit = { inputStreamReader = new InputStreamReader(inputStream) - BufferedReader = new BufferedReader(inputStreamReader) + bufferedReader = new BufferedReader(inputStreamReader) } @scala.throws[IOException] @@ -82,25 +79,26 @@ class StorageScriptFsReader(pathP: FsPath, charsetP: String, inputStreamP: Input override def available: Long = if (inputStream != null) inputStream.available() else 0L override def close(): Unit = { - if (BufferedReader != null) BufferedReader.close() - if (inputStreamReader != null) inputStreamReader.close() - if (inputStream != null) inputStream.close() + IOUtils.closeQuietly(bufferedReader) + IOUtils.closeQuietly(inputStreamReader) + IOUtils.closeQuietly(inputStream) } /** * Determine if the read line is metadata(判断读的行是否是metadata) + * * @param line * @return */ - def isMetadata(line: String, prefix: String,prefixConf:String): Boolean = { + def isMetadata(line: String, prefix: String, prefixConf: String): Boolean = { val regex = ("\\s*" + prefix + "\\s*(.+)\\s*" + "=" + "\\s*(.+)\\s*").r line match { - case regex(_,_) => true + case regex(_, _) => true case _ => { val split: Array[String] = line.split("=") - if(split.size !=2) return false - if (split(0).split(" ").filter(_!="").size != 4) return false - if(!split(0).split(" ").filter(_!="")(0).equals(prefixConf)) return false + if (split.size != 2) return false + if (split(0).split(" ").filter(_ != "").size != 4) return false + if (!split(0).split(" ").filter(_ != "")(0).equals(prefixConf)) return false true } } diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/writer/StorageScriptFsWriter.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/writer/StorageScriptFsWriter.scala index 8f223ebc0d..d506944b10 100644 --- a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/writer/StorageScriptFsWriter.scala +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/script/writer/StorageScriptFsWriter.scala @@ -16,43 +16,57 @@ package com.webank.wedatasphere.linkis.storage.script.writer -import java.io.{IOException, OutputStream} +import java.io.{ByteArrayInputStream, IOException, InputStream, OutputStream} import java.util import com.webank.wedatasphere.linkis.common.io.{FsPath, MetaData, Record} -import com.webank.wedatasphere.linkis.storage.script.{Compaction, ScriptFsWriter, ScriptMetaData, ScriptRecord} -import com.webank.wedatasphere.linkis.storage.utils.StorageUtils +import com.webank.wedatasphere.linkis.storage.LineRecord +import com.webank.wedatasphere.linkis.storage.script.{Compaction, ScriptFsWriter, ScriptMetaData} +import com.webank.wedatasphere.linkis.storage.utils.{StorageConfiguration, StorageUtils} import org.apache.commons.io.IOUtils /** * Created by johnnwang on 2018/10/23. */ -class StorageScriptFsWriter(pathP: FsPath, charsetP: String, outputStreamP: OutputStream) extends ScriptFsWriter { - override val path: FsPath = pathP - override val charset: String = charsetP - override val outputStream: OutputStream = outputStreamP +class StorageScriptFsWriter(val path: FsPath, val charset: String, outputStream: OutputStream = null) extends ScriptFsWriter { + + private val stringBuilder = new StringBuilder @scala.throws[IOException] override def addMetaData(metaData: MetaData): Unit = { val compactions = Compaction.listCompactions().filter(p => p.belongTo(StorageUtils.pathToSuffix(path.getPath))) - val metadataLineJ = new util.ArrayList[String]() + val metadataLine = new util.ArrayList[String]() if (compactions.length > 0) { - metaData.asInstanceOf[ScriptMetaData].getMetaData.map(compactions(0).compact).foreach(metadataLineJ.add) - IOUtils.writeLines(metadataLineJ, "\n", outputStream, charset) + metaData.asInstanceOf[ScriptMetaData].getMetaData.map(compactions(0).compact).foreach(metadataLine.add) + if (outputStream != null) { + IOUtils.writeLines(metadataLine, "\n", outputStream, charset) + } else { + import scala.collection.JavaConversions._ + metadataLine.foreach(m => stringBuilder.append(s"$m\n")) + } } } @scala.throws[IOException] override def addRecord(record: Record): Unit = { - val scriptRecord = record.asInstanceOf[ScriptRecord] - IOUtils.write(scriptRecord.getLine, outputStream, charset) + //转成LineRecord而不是TableRecord是为了兼容非Table类型的结果集写到本类中 + val scriptRecord = record.asInstanceOf[LineRecord] + if (outputStream != null) { + IOUtils.write(scriptRecord.getLine, outputStream, charset) + } else { + stringBuilder.append(scriptRecord.getLine) + } } override def close(): Unit = { - StorageUtils.close(outputStream) + IOUtils.closeQuietly(outputStream) } override def flush(): Unit = if (outputStream != null) outputStream.flush() + def getInputStream(): InputStream = { + new ByteArrayInputStream(stringBuilder.toString().getBytes(StorageConfiguration.STORAGE_RS_FILE_TYPE.getValue)) + } + } diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/source/AbstractFileSource.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/source/AbstractFileSource.scala new file mode 100644 index 0000000000..0d08318dec --- /dev/null +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/source/AbstractFileSource.scala @@ -0,0 +1,136 @@ +/* + * Copyright 2019 WeBank + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.webank.wedatasphere.linkis.storage.source + +import java.util + +import com.webank.wedatasphere.linkis.common.io.{FsReader, FsWriter, MetaData, Record} +import com.webank.wedatasphere.linkis.storage.resultset.table.{TableMetaData, TableRecord} +import com.webank.wedatasphere.linkis.storage.script.{ScriptMetaData, VariableParser} +import com.webank.wedatasphere.linkis.storage.{LineMetaData, LineRecord} +import org.apache.commons.io.IOUtils +import org.apache.commons.math3.util.Pair + +import scala.collection.JavaConversions._ + +/** + * Created by patinousward on 2020/1/15. + */ +abstract class AbstractFileSource extends FileSource { + + protected var start: Int = 0 + + protected var end: Int = -1 + + protected var count: Int = 0 + + protected var totalLine = 0 + + protected var shuffler: Record => Record = r => r + + protected var pageTrigger: Boolean = false + + protected var params: util.Map[String, String] = new util.HashMap[String, String] + + protected var fsReader: FsReader[_ <: MetaData, _ <: Record] = _ + + protected var `type`: String = "script/text" + + def setType(`type`: String): AbstractFileSource = { + params += "type" -> `type` + this.`type` = `type` + this + } + + override def shuffle(s: Record => Record): AbstractFileSource = { + this.shuffler = s + this + } + + override def page(page: Int, pageSize: Int): AbstractFileSource = { + if (pageTrigger) return this + start = (page - 1) * pageSize + end = pageSize * page - 1 + pageTrigger = true + this + } + + override def addParams(params: util.Map[String, String]): AbstractFileSource = { + this.params.putAll(params) + this + } + + override def addParams(key: String, value: String): FileSource = { + this.params += key -> value + this + } + + override def getParams(): util.Map[String, String] = this.params + + def setFsReader[K <: MetaData, V <: Record](fsReader: FsReader[K, V]): AbstractFileSource = { + this.fsReader = fsReader + this + } + + override def write[K <: MetaData, V <: Record](fsWriter: FsWriter[K, V]): Unit = { + `while`(fsWriter.addMetaData, fsWriter.addRecord) + } + + def `while`[M](m: MetaData => M, r: Record => Unit): M = { + val metaData = fsReader.getMetaData + val t = m(metaData) + while (fsReader.hasNext && ifContinueRead) { + if (ifStartRead) { + r(shuffler(fsReader.getRecord)) + totalLine += 1 + } + count += 1 + } + t + } + + override def close(): Unit = IOUtils.closeQuietly(this.fsReader) + + def collectRecord(record: Record): Array[String] = { + record match { + case t: TableRecord => t.row.map(_.toString) + case l: LineRecord => Array(l.getLine) + } + } + + override def collect(): Pair[Object, util.ArrayList[Array[String]]] = { + val record = new util.ArrayList[Array[String]] + val metaData = `while`(collectMetaData, r => record.add(collectRecord(r))) + this.params += "totalLine" -> totalLine.toString + new Pair(metaData, record) + } + + //如果不分页,则一直读,如果分页,则 count需要小于count + def ifContinueRead: Boolean = !pageTrigger || count <= end + + def ifStartRead: Boolean = !pageTrigger || count >= start + + def collectMetaData(metaData: MetaData): Object = { + //script/text ,tableResultset,lineResultSet + metaData match { + case s: ScriptMetaData => VariableParser.getMap(s.getMetaData) + case l: LineMetaData => l.getMetaData + case t: TableMetaData => t.columns.map(_.toString) + } + } + +} diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/source/FileSource.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/source/FileSource.scala new file mode 100644 index 0000000000..9540b16515 --- /dev/null +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/source/FileSource.scala @@ -0,0 +1,86 @@ +/* + * Copyright 2019 WeBank + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.webank.wedatasphere.linkis.storage.source + +import java.io.{Closeable, InputStream} +import java.util + +import com.webank.wedatasphere.linkis.common.io._ +import com.webank.wedatasphere.linkis.storage.exception.StorageErrorException +import com.webank.wedatasphere.linkis.storage.resultset.{ResultSetFactory, ResultSetReader} +import com.webank.wedatasphere.linkis.storage.script.ScriptFsReader +import com.webank.wedatasphere.linkis.storage.utils.StorageConfiguration +import org.apache.commons.math3.util.Pair + +/** + * Created by patinousward on 2020/1/15. + */ +trait FileSource extends Closeable { + + def shuffle(s: Record => Record): FileSource + + def page(page: Int, pageSize: Int): FileSource + + def collect(): Pair[Object, util.ArrayList[Array[String]]] + + def write[K <: MetaData, V <: Record](fsWriter: FsWriter[K, V]): Unit + + def addParams(params: util.Map[String, String]): FileSource + + def addParams(key: String, value: String): FileSource + + def getParams(): util.Map[String, String] + +} + +object FileSource { + + private val fileType = Array("dolphin", "sql", "scala", "py", "hql", "python", "out", "log", "text", "sh", "jdbc", "mlsql") + + private val suffixPredicate = (path: String, suffix: String) => path.endsWith(s".$suffix") + + def isResultSet(path: String): Boolean = { + suffixPredicate(path, fileType.head) + } + + def isTableResultSet(fileSource: FileSource): Boolean = { + ResultSetFactory.TABLE_TYPE.equals(fileSource.getParams().get("type")) + } + + def create(fsPath: FsPath, fs: Fs): FileSource = { + create(fsPath, fs.read(fsPath)) + } + + def create(fsPath: FsPath, is: InputStream): FileSource = { + canRead(fsPath.getPath) + if (isResultSet(fsPath.getPath)) { + val resultset = ResultSetFactory.getInstance.getResultSetByPath(fsPath) + val resultsetReader = ResultSetReader.getResultSetReader(resultset, is) + new ResultsetFileSource().setFsReader(resultsetReader).setType(resultset.resultSetType()) + } else { + val scriptFsReader = ScriptFsReader.getScriptFsReader(fsPath, StorageConfiguration.STORAGE_RS_FILE_TYPE.getValue, is) + new TextFileSource().setFsReader(scriptFsReader) + } + } + + private def canRead(path: String) = { + if (!fileType.exists(suffixPredicate(path, _))) throw new StorageErrorException(54001, "不支持打开的文件类型") + } + +} + + diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/csv/CSVFsWriter.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/source/ResultsetFileSource.scala similarity index 52% copy from storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/csv/CSVFsWriter.scala copy to storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/source/ResultsetFileSource.scala index 66dc4b123d..a82cb0b2ba 100644 --- a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/csv/CSVFsWriter.scala +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/source/ResultsetFileSource.scala @@ -14,26 +14,23 @@ * limitations under the License. */ -package com.webank.wedatasphere.linkis.storage.csv +package com.webank.wedatasphere.linkis.storage.source -import java.io.InputStream - -import com.webank.wedatasphere.linkis.common.io.FsWriter +import com.webank.wedatasphere.linkis.storage.resultset.table.TableRecord +import com.webank.wedatasphere.linkis.storage.utils.StorageUtils /** - * Created by johnnwang on 2018/11/12. + * Created by patinousward on 2020/1/15. */ -abstract class CSVFsWriter extends FsWriter { - val charset: String - val separator: String - var isLastRow: Boolean = false - - def setIsLastRow(value: Boolean): Unit +class ResultsetFileSource extends AbstractFileSource { - def getCSVStream: InputStream -} + shuffler = { + case t: TableRecord => new TableRecord(t.row.map { + case null => params.getOrDefault("nullValue", "NULL") + case value: Double => StorageUtils.doubleToString(value) + case r => r + }) + case record => record + } -object CSVFsWriter { - def getCSVFSWriter(charset: String, separator: String): CSVFsWriter = new StorageCSVWriter(charset, separator) } - diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/source/TextFileSource.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/source/TextFileSource.scala new file mode 100644 index 0000000000..01f5fef1c9 --- /dev/null +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/source/TextFileSource.scala @@ -0,0 +1,52 @@ +/* + * Copyright 2019 WeBank + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.webank.wedatasphere.linkis.storage.source + +import java.util + +import com.webank.wedatasphere.linkis.storage.LineRecord +import com.webank.wedatasphere.linkis.storage.script.ScriptRecord +import org.apache.commons.math3.util.Pair + +/** + * Created by patinousward on 2020/1/15. + */ +class TextFileSource extends AbstractFileSource { + + shuffler = { + case s: ScriptRecord if "".equals(s.getLine) => new LineRecord("\n") + case record => record + } + + override def collect(): Pair[Object, util.ArrayList[Array[String]]] = { + val collect: Pair[Object, util.ArrayList[Array[String]]] = super.collect() + if (!params.getOrDefault("ifMerge", "true").toBoolean) return collect + import scala.collection.JavaConversions._ + val snd: util.ArrayList[Array[String]] = collect.getSecond + val str = new StringBuilder + for (i <- snd) { + i match { + case Array("\n") => str.append("\n") + case Array(y) => str.append(y).append("\n") + } + } + snd.clear() + snd.add(Array(str.toString())) + collect + } + +} diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/utils/StorageConfiguration.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/utils/StorageConfiguration.scala index 380d0f9dfc..c49ae2af18 100644 --- a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/utils/StorageConfiguration.scala +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/utils/StorageConfiguration.scala @@ -16,7 +16,7 @@ package com.webank.wedatasphere.linkis.storage.utils -import com.webank.wedatasphere.linkis.common.conf.{ByteType, CommonVars, Configuration} +import com.webank.wedatasphere.linkis.common.conf.{ByteType, CommonVars} /** * Created by johnnwang on 10/15/18. @@ -29,7 +29,7 @@ object StorageConfiguration { val HDFS_ROOT_USER = CommonVars("wds.linkis.storage.hdfs.root.user", "hadoop") - val LOCAL_ROOT_USER = CommonVars("wds.linkis.storage.hdfs.root.user", "root") + val LOCAL_ROOT_USER = CommonVars("wds.linkis.storage.local.root.user", "root") val STORAGE_USER_GROUP = CommonVars("wds.linkis.storage.fileSystem.group", "bdap") @@ -57,4 +57,6 @@ object StorageConfiguration { val IO_INIT_RETRY_LIMIT = CommonVars("wds.linkis.storage.io.init.retry.limit", 10) val STORAGE_HDFS_GROUP = CommonVars("wds.linkis.storage.fileSystem.hdfs.group", "hadoop") + + val DOUBLE_FRACTION_LEN = CommonVars[Int]("wds.linkis.double.fraction.length", 30) } diff --git a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/utils/StorageUtils.scala b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/utils/StorageUtils.scala index b3ae098830..5a57a051f5 100644 --- a/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/utils/StorageUtils.scala +++ b/storage/storage/src/main/scala/com/webank/wedatasphere/linkis/storage/utils/StorageUtils.scala @@ -18,13 +18,14 @@ package com.webank.wedatasphere.linkis.storage.utils import java.io.{Closeable, File, InputStream, OutputStream} import java.lang.reflect.Method +import java.text.NumberFormat import com.webank.wedatasphere.linkis.common.conf.Configuration import com.webank.wedatasphere.linkis.common.io.{Fs, FsPath} import com.webank.wedatasphere.linkis.common.utils.{Logging, Utils} import com.webank.wedatasphere.linkis.storage.exception.StorageFatalException -import com.webank.wedatasphere.linkis.storage.{LineMetaData, LineRecord} import com.webank.wedatasphere.linkis.storage.resultset.{ResultSetFactory, ResultSetReader, ResultSetWriter} +import com.webank.wedatasphere.linkis.storage.{LineMetaData, LineRecord} import org.apache.commons.lang.StringUtils import scala.collection.mutable @@ -41,6 +42,13 @@ object StorageUtils extends Logging{ val FILE_SCHEMA = "file://" val HDFS_SCHEMA = "hdfs://" + private val nf = NumberFormat.getInstance() + nf.setGroupingUsed(false) + nf.setMaximumFractionDigits(StorageConfiguration.DOUBLE_FRACTION_LEN.getValue) + + def doubleToString(value:Double): String ={ + nf.format(value) + } def loadClass[T](classStr: String, op: T => String): Map[String, T] = { val _classes = classStr.split(",") --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
