package com.jasongj.spark.writer;

import com.jasongj.spark.model.DataType;
import com.jasongj.spark.model.TableMetaData;
import com.jasongj.spark.model.Tuple;
import org.apache.commons.compress.utils.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.api.TableMeta;

import java.io.DataOutputStream;
import java.io.IOException;
import java.sql.Date;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;

/**
 * Created by Jason Guo (jason.guo.vip@gmail.com)
 */
public class TextTupleWriter extends TupleWriter {

    private SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
    private SimpleDateFormat timestampFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    private String fieldDelimiter, lineDelimiter;

    public TextTupleWriter(DataOutputStream dataOutputStream, Configuration hadoopConfiguration, Path path, TableMetaData outputTableMetaData) {
        super(dataOutputStream, hadoopConfiguration, path, outputTableMetaData);
        this.fieldDelimiter = String.valueOf(outputTableMetaData.getFieldDelimiter());
        this.lineDelimiter = String.valueOf(outputTableMetaData.getLineDelimiter());
    }

    @Override
    public boolean init() {
        return super.init() && outputTableMetaData.getDataType() == DataType.TEXT;
    }

    @Override
    public void write(Tuple tuple) {
        String data = StringUtils.join(tuple.getData().stream().map((Object object) -> {
            if(object == null) {
                return StringUtils.EMPTY;
            } else if(object instanceof Date) {
                return dateFormat.format((Date) object);
            } else if(object instanceof Timestamp) {
                return timestampFormat.format((Timestamp) object);
            } else {
                return object.toString();
            }
        }).toArray(), fieldDelimiter) + lineDelimiter;
        try {
            dataOutputStream.write(data.getBytes());
        } catch (IOException ex) {
            throw new RuntimeException("Write Tuple with TextTupleWriter exception", ex);
        }
    }

    @Override
    public void close() {
        if(dataOutputStream != null) {
            IOUtils.closeQuietly(dataOutputStream);
            dataOutputStream = null;
        }
    }
}

@@ -183,22 +183,10 @@
 ons.
-compress.utils
+io
 .IOU
@@ -308,20 +308,34 @@
 doop.fs.
-Path
+FSDataOutputStream
 ;%0Aimport
@@ -357,71 +357,130 @@
 oop.
-hive.metastore.api.TableMeta;%0A%0Aimport java.io.DataOutputStream;
+fs.FileSystem;%0Aimport org.apache.hadoop.fs.Path;%0Aimport parquet.org.slf4j.Logger;%0Aimport parquet.org.slf4j.LoggerFactory;%0A
 %0Aimp
@@ -697,16 +697,102 @@
 Writer %7B
+%0A    private static final Logger LOG = LoggerFactory.getLogger(TextTupleWriter.class);
 %0A%0A    pr
@@ -1009,36 +1009,22 @@
 er;%0A
-%0A
     p
-ublic TextTupleWriter(
+rivate FS
 Data
@@ -1052,18 +1052,46 @@
 utStream
-, 
+;%0A%0A    public TextTupleWriter(
 Configur
@@ -1183,26 +1183,8 @@
 per(
-dataOutputStream, 
 hado
@@ -1457,23 +1457,19 @@
         
-return 
+if(
 super.in
@@ -1526,17 +1526,446 @@
 ype.TEXT
-;
+) %7B%0A            FileSystem fileSystem = null;%0A            try %7B%0A                fileSystem = FileSystem.newInstance(hadoopConfiguration);%0A                dataOutputStream = fileSystem.create(path, true);%0A                return true;%0A            %7D catch (IOException ex) %7B%0A                LOG.error(%22Init TextTupleWriter failed%22, ex);%0A                return false;%0A            %7D%0A        %7D else %7B%0A            return false;%0A        %7D
 %0A    %7D%0A%0A
@@ -2797,142 +2797,53 @@
     
-if(dataOutputStream != null) %7B%0A            IOUtils.closeQuietly(dataOutputStream);%0A            dataOutputStream = null;%0A    
+IOUtils.closeQuietly(dataOutputStream);%0A
     %7D%0A
-    %7D
 %0A%7D%0A

refactor TextTupleWriter -- create file in init method