spark/src/DataStreamSuite/DataStream/Primitives/SideEffectStreams/CSVSaverStream.cs

using System;
using System.Collections.Generic;
using System.Text;
using System.IO;
using BCG.Utilities;
using System.Diagnostics;
using Microsoft.Data.Analysis;
using Parquet;
using Parquet.Data;


namespace BCG {
   [DataStreamUserRespresentation("CSVSaver")]
   public class CSVSaverStream : ViewStream, ISideEffectStream {
      private IBlueTableSchema myResultSchema = new BlueTableSchema("CSVResult", new ColumnInfo("id", 0, typeof(int)),
         new ColumnInfo("fileName", 1, typeof(string)),
         new ColumnInfo("nWrites", 2, typeof(int)),
         new ColumnInfo("nLastRows", 3, typeof(int)));

      private string myFullyQualifiedFileName;
      private ViewStream myStream;
      private bool myOverwrite, myFireResultSnapshot;
      private int myNWrites = 0, myNMaxWrites = Int32.MaxValue;

      public CSVSaverStream(ViewStream vs, string fullyQualifiedFileName)
         : this(vs, fullyQualifiedFileName, true, Int32.MaxValue, false) { }

      public CSVSaverStream(ViewStream vs, string fullyQualifiedFileName, bool overwrite, int nMaxWrites, bool fireResultSnapshot) {
         myStream = vs;
         myFullyQualifiedFileName = fullyQualifiedFileName;
         myOverwrite = overwrite;
         myNMaxWrites = nMaxWrites;
         myFireResultSnapshot = fireResultSnapshot;
      }

      protected override void myOnFirstSubscriber() { myStream.onChange += new ChangeDelegate(myHandleChanges); }

      protected override void myOnLastSubscriber() { myStream.onChange -= new ChangeDelegate(myHandleChanges); }

      private void myHandleChanges(object newValue) {
         IBlueRowCollection totalState = (newValue as DataPacket).newRows;
         if (totalState.Count > 0 && myNWrites < myNMaxWrites) {
            myNWrites++;
            int i = 0;
            StreamWriter writer = new StreamWriter(new FileStream(myGetNewFileName(), FileMode.Create, FileAccess.Write));
            foreach (IBlueRow r in totalState) {
               if (i++ == 0) {
                  string header = String.Empty;
                  foreach (string col in r.schema.columnNames)
                     if (col != "id")
                        header += (header == String.Empty ? col : "," + col);
                  writer.WriteLine(header);
                  writer.Flush();
               }
               writer.WriteLine(myGetCSVRow(r));
               writer.Flush();
            }
            writer.Close();
            if (myFireResultSnapshot) {
               IBlueRow nr = new BlueRow(myResultSchema, 0, myFullyQualifiedFileName, myNWrites, totalState.Count);
               BlueTable nRows = new BlueTable(myResultSchema);
               nRows.addRow(nr);
               myFireEvent(new DataPacket(nRows, nRows, new BlueTable(myLastValue == null ? new BlueTable(myResultSchema) : (myLastValue as DataPacket).newRows)));
            } else
               myFireEvent(newValue);
         }
      }

      private string myGetNewFileName() {
         string[] tags = myFullyQualifiedFileName.Split('.');
         if (myOverwrite)
            return tags[0] + ".csv";
         else
            return tags[0] + "_" + myFormateDateTime(DateTime.Now) + ".csv";
      }

      private static string myGetCSVRow(IBlueRow r) {
         string rv = String.Empty;
         int i = 0;
         foreach (string col in r.schema.columnNames) {
            if (col != "id") {
               string colText = String.Empty;
               if (r[col] != null && r[col] != DBNull.Value)
                  colText = r[col].ToString();
               if (r.schema[col].colType == typeof(string))
                  colText = "\"" + colText + "\"";
               rv += (i++ == 0 ? colText : "," + colText);
            }
         }
         return rv;
      }

      private string myFormateDateTime(DateTime dt) {
         return
         string.Format("{0}-{1}-{2}-{3}-{4}-{5}-{6}",
                        dt.Year.ToString("0000"),
                        dt.Month.ToString("00"),
                        dt.Day.ToString("00"),
                        dt.Hour.ToString("00"),
                        dt.Minute.ToString("00"),
                        dt.Second.ToString("00"),
                        dt.Millisecond.ToString("0000"));
      }
   }

   [DataStreamUserRespresentation("CSVReader")]
   public class CSVReaderStream : ViewStream {
      private readonly string _file_address;
      Stopwatch _stopwatch = new Stopwatch();
      public CSVReaderStream(string fileAddress) {
         _file_address = fileAddress;
      }

      protected override void myOnFirstSubscriber() {
         _stopwatch.Start();
         DataFrame df = DataFrame.LoadCsv(_file_address);
         _stopwatch.Stop();
         bts.Log.Gen.info($"CSVReaderStream.myOnFirstSubscriber stream csv load col_count: {df.Columns.Count} rows, time: {_stopwatch.ElapsedMilliseconds} msecs");
         _stopwatch.Start();
         IBlueRowCollection rows = BlueDataUtility.createRowsFromCSVFileWithoutTypes(_file_address);
         _stopwatch.Stop();
         bts.Log.Gen.info($"CSVReaderStream.myOnFirstSubscriber stream csv load count: {rows.Count} rows, time: {_stopwatch.ElapsedMilliseconds} msecs");
         DataPacket dp_out = BlueDataUtility.build_dp_inc(rows, myLastValue as DataPacket);
         myFireEvent(dp_out);
      }
   }

   [DataStreamUserRespresentation("CSVReaderStreamMapper")]
   public class CSVReaderStreamMapperStream : ViewStream {
      public CSVReaderStreamMapperStream(ViewStream stream, string output_schema) {
         mySetRealSource(new MapStream(stream, new CSVReaderStreamMapper(output_schema)));
      }
   }

   public class CSVReaderStreamMapper : IStreamMapper {
      private IBlueTableSchema mySchema;
      private readonly string[] myOutputSchemaStringArray;

      public CSVReaderStreamMapper(string outputSchema) {
         myOutputSchemaStringArray = outputSchema.Split(' ');
      }

      public IBlueTableSchema getOutputSchema(IBlueTableSchema inputSchema) {
         if (mySchema == null) {
            ColumnInfo[] ci = inputSchema.columnInfo;
            foreach (string col in myOutputSchemaStringArray) {
               string colName = col.Substring(0, col.IndexOf('\\')).ToLower();
               for (int i = 0; i < ci.Length; i++) {
                  if (ci[i].colName.ToLower().Equals(colName)) {
                     Type t = myGetType(col.Substring(col.IndexOf('\\') + 1));
                     ci[i] = new ColumnInfo(ci[i].colName, i, t);
                     break;
                  }
               }
            }
            mySchema = new BlueTableSchema(ci, "CSVReader");
         }
         return mySchema;
      }

      public IBlueRow map(IBlueRow row) {
         object[] vals = new object[row.schema.columnInfo.Length];
         row.Values.CopyTo(vals, 0);
         for (int i = 0; i < mySchema.columnInfo.Length; i++) {
            Type t = mySchema.columnInfo[i].colType;
            try {
               vals[i] = System.Convert.ChangeType(vals[i], mySchema.columnInfo[i].colType, System.Globalization.CultureInfo.InvariantCulture);
            } catch {
               if (t.Equals(typeof(Int16))) {
                  Int16 result;
                  if (Int16.TryParse((string)vals[i], out result)) {
                     vals[i] = result;
                     continue;
                  }
                  double r2;
                  if (Double.TryParse((string)vals[i], out r2)) {
                     try {
                        vals[i] = (Int16)r2;
                        continue;
                     } catch { }
                  }
               } else if (t.Equals(typeof(Int32))) {
                  Int32 result;
                  if (Int32.TryParse((string)vals[i], out result)) {
                     vals[i] = result;
                     continue;
                  }
                  double r2;
                  if (Double.TryParse((string)vals[i], out r2)) {
                     try {
                        vals[i] = (Int32)r2;
                        continue;
                     } catch { }
                  }
               } else if (t.Equals(typeof(Int64))) {
                  Int64 result;
                  if (Int64.TryParse((string)vals[i], out result)) {
                     vals[i] = result;
                     continue;
                  }
                  double r2;
                  if (Double.TryParse((string)vals[i], out r2)) {
                     try {
                        vals[i] = (Int64)r2;
                        continue;
                     } catch { }
                  }
               }
               vals[i] = DBNull.Value;
            }
         }
         return new BlueRow(mySchema, vals);
      }

      private Type myGetType(string type) {
         type = type.ToLower();
         if (type.Contains("int"))
            return typeof(int);
         else if (type.Contains("double"))
            return typeof(double);
         else if (type.Contains("date"))
            return typeof(DateTime);
         else
            return typeof(string);
      }
   }
}