From 99a5eb31ba8cd686c301c2f39e724f398f42c9f0 Mon Sep 17 00:00:00 2001 From: Michael Sharp Date: Fri, 7 Feb 2020 14:10:04 -0800 Subject: [PATCH 1/3] added in DateTime type support for TSI --- .../TimeSeriesImputer.cs | 57 +++++++++- .../TimeSeriesImputerDataView.cs | 57 ++++++++++ .../Transformers/TimeSeriesImputerTests.cs | 100 +++++++++++++++++- 3 files changed, 210 insertions(+), 4 deletions(-) diff --git a/src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs b/src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs index 537f0611b9..00462d649e 100644 --- a/src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs +++ b/src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs @@ -32,7 +32,7 @@ public static class TimeSeriesImputerExtensionClass /// purpose of this estimator. Other column types will have the default value placed if a row is imputed. /// /// The transform catalog. - /// Column representing the time series. Should be of type + /// Column representing the time series. Should be of type or /// List of columns to use as grains /// Mode of imputation for missing values in column. If not passed defaults to forward fill public static TimeSeriesImputerEstimator ReplaceMissingTimeSeriesValues(this TransformsCatalog catalog, string timeSeriesColumn, string[] grainColumns, @@ -46,7 +46,7 @@ public static TimeSeriesImputerEstimator ReplaceMissingTimeSeriesValues(this Tra /// purpose of this estimator. /// /// The transform catalog. - /// Column representing the time series. Should be of type + /// Column representing the time series. Should be of type or /// List of columns to use as grains /// List of columns to filter. If is than columns in the list will be ignored. /// If is than values in the list are the only columns imputed. @@ -83,6 +83,16 @@ public static TimeSeriesImputerEstimator ReplaceMissingTimeSeriesValues(this Tra /// | Output column data type | All Types | /// | Exportable to ONNX | No | /// + /// The TimeSeriesImputer imputes missing rows and column data per grain (category), based on the dates in the date column. This operation needs to happen to every column in the IDataView, + /// If you "filter" a column using the filterColumns and filterMode parameters, if a row is imputed the default value for that type will be used. + /// Currently only float/double/string columns are supported for imputation strategies, and an empty string is considered "missing" for the + /// purpose of this estimator. A new column is added to the schema after this operation is run. The column is called "IsRowImputed" and is a + /// boolean value representing if the row was created as a result of this operation or not. + /// + /// NOTE: It is not recommended to chain this multiple times. If a column is filtered, the default value is placed when a row is imputed, and the + /// default value is not null. Thus any other TimeSeriesImputers will not be able to replace those values anymore causing essentially a very + /// computationally expensive NO-OP. + /// /// The is not a trivial estimator and needs training. /// /// @@ -98,7 +108,7 @@ public sealed class TimeSeriesImputerEstimator : IEstimator _currentSupportedTypes = new List { typeof(sbyte), typeof(byte), typeof(short), typeof(ushort), typeof(int), typeof(uint), - typeof(long), typeof(ulong), typeof(float), typeof(double), typeof(string), typeof(ReadOnlyMemory)}; + typeof(long), typeof(ulong), typeof(float), typeof(double), typeof(string), typeof(ReadOnlyMemory), typeof(DateTime)}; #region Options internal sealed class Options : TransformInputBase @@ -127,6 +137,12 @@ internal sealed class Options : TransformInputBase #region Class Enums + /// + /// This is the representation of which Imputation Strategy to use. + /// ForwardFill takes the value from the last good row and propagates it forward anytime a row is imputer or a missing value is found. + /// BackFill is the same as ForwardFill, expect it takes from the next good row and propagates backwards. + /// Median only supports float/double, takes the median value found during training and uses that to replace missing values + /// public enum ImputationStrategy : byte { ForwardFill = 1, @@ -135,6 +151,12 @@ public enum ImputationStrategy : byte // Interpolate = 4, interpolate not currently supported in the native code. }; + /// + /// What the filter strategy used is. + /// NoFilter takes all of the columns so you dont have to specify anything. + /// Include only does the specified ImputationStrategy on the columns you specify. The other columns will get a default value. + /// Exclude is the exact opposite of Include, and does the ImputationStrategy on all columns but the ones you specify, which will get the default value. + /// public enum FilterMode : byte { NoFilter = 1, @@ -559,6 +581,8 @@ internal static TypedColumn CreateTypedColumn(DataViewSchema.Column column, stri return new NumericTypedColumn(column, optionalColumns.Contains(column.Name)); else if (type == typeof(ReadOnlyMemory).ToString()) return new StringTypedColumn(column, optionalColumns.Contains(column.Name)); + else if (type == typeof(DateTime).ToString()) + return new DateTimeTypedColumn(column, optionalColumns.Contains(column.Name)); throw new InvalidOperationException($"Unsupported type {type}"); } @@ -637,6 +661,33 @@ internal override byte[] GetSerializedValue() } } + private class DateTimeTypedColumn : TypedColumn + { + private static readonly DateTime _unixEpoch = new DateTime(1970, 1, 1); + private readonly bool _isNullable; + + internal DateTimeTypedColumn(DataViewSchema.Column column, bool isNullable = false) : + base(column) + { + _isNullable = isNullable; + } + + internal override byte[] GetSerializedValue() + { + var dateTime = GetValue(); + byte[] bytes; + + var value = dateTime.Subtract(_unixEpoch).Ticks / TimeSpan.TicksPerSecond; + + bytes = BitConverter.GetBytes(value); + + if (_isNullable) + return new byte[1] { Convert.ToByte(true) }.Concat(bytes).ToArray(); + else + return bytes; + } + } + #endregion } diff --git a/src/Microsoft.ML.Featurizers/TimeSeriesImputerDataView.cs b/src/Microsoft.ML.Featurizers/TimeSeriesImputerDataView.cs index ccb33b99e9..d7baff84ae 100644 --- a/src/Microsoft.ML.Featurizers/TimeSeriesImputerDataView.cs +++ b/src/Microsoft.ML.Featurizers/TimeSeriesImputerDataView.cs @@ -103,6 +103,8 @@ internal static TypedColumn CreateTypedColumn(DataViewSchema.Column column, stri return new StringTypedColumn(column, optionalColumns.Contains(column.Name), allImputedColumns.Contains(column.Name), state); else if (type == typeof(bool).ToString()) return new BoolTypedColumn(column, optionalColumns.Contains(column.Name), allImputedColumns.Contains(column.Name), state); + else if (type == typeof(DateTime).ToString()) + return new DateTimeTypedColumn(column, optionalColumns.Contains(column.Name), allImputedColumns.Contains(column.Name), state); throw new InvalidOperationException($"Unsupported type {type}"); } @@ -586,6 +588,61 @@ internal override unsafe int GetDataSizeInBytes(byte* data, int currentOffset) } } + private class DateTimeTypedColumn : TypedColumn + { + private static readonly DateTime _unixEpoch = new DateTime(1970, 1, 1); + private readonly bool _isNullable; + + internal DateTimeTypedColumn(DataViewSchema.Column column, bool isNullable, bool isImputed, SharedColumnState state) : + base(column, isImputed, state) + { + _isNullable = isNullable; + } + + internal override byte[] GetSerializedValue() + { + var dateTime = GetSourceValue(); + byte[] bytes; + + var value = dateTime.Subtract(_unixEpoch).Ticks / TimeSpan.TicksPerSecond; + + bytes = BitConverter.GetBytes(value); + + if (_isNullable) + return new byte[1] { Convert.ToByte(true) }.Concat(bytes).ToArray(); + else + return bytes; + } + + internal unsafe override DateTime GetDataFromNativeBinaryArchiveData(byte* data, int offset) + { + long value; + if (_isNullable) + { + if (!BoolTypedColumn.GetBoolFromNativeBinaryArchiveData(data, offset)) // If value not present return empty string + return new DateTime(); + + value = *(long*)(data + offset + 1); // Add 1 for the byte bool flag + + } + else + { + value = *(long*)(data + offset); + } + + return new DateTime(_unixEpoch.Ticks + (value * TimeSpan.TicksPerSecond)); + + } + + internal override unsafe int GetDataSizeInBytes(byte* data, int currentOffset) + { + if (_isNullable) + return 1 + sizeof(long); // + 1 for the byte bool flag + + return sizeof(long); + } + } + #endregion #region Native Exports diff --git a/test/Microsoft.ML.Tests/Transformers/TimeSeriesImputerTests.cs b/test/Microsoft.ML.Tests/Transformers/TimeSeriesImputerTests.cs index bed0a67e5f..013c59f6cc 100644 --- a/test/Microsoft.ML.Tests/Transformers/TimeSeriesImputerTests.cs +++ b/test/Microsoft.ML.Tests/Transformers/TimeSeriesImputerTests.cs @@ -190,6 +190,61 @@ public void Forwardfill() Done(); } + [NotCentOS7Fact] + public void DateTimeSupportForwardfill() + { + MLContext mlContext = new MLContext(1); + var dataList = new[] { new { date = new DateTime(1970, 1, 1), grainA = "A", dataA = 2.0f }, + new { date = new DateTime(1970, 1, 3), grainA = "A", dataA = float.NaN }, + new { date = new DateTime(1970, 1, 5), grainA = "A", dataA = 5.0f }, + new { date = new DateTime(1970, 1, 7), grainA = "A", dataA = float.NaN }, + new { date = new DateTime(1970, 1, 8), grainA = "A", dataA = float.NaN }}; + var data = mlContext.Data.LoadFromEnumerable(dataList); + + // Build the pipeline, fit, and transform it. + var pipeline = mlContext.Transforms.ReplaceMissingTimeSeriesValues("date", new string[] { "grainA" }); + var model = pipeline.Fit(data); + var output = model.Transform(data); + var prev = output.Preview(); + + // Should have 3 original columns + 1 more for IsRowImputed + Assert.Equal(4, output.Schema.Count); + + // Imputing rows with days for 2,4,6, so should have length of 8 + Assert.Equal(8, prev.RowView.Length); + + // Check that imputed rows have the correct dates + Assert.Equal(new DateTime(1970, 1, 2), prev.ColumnView[0].Values[1]); + Assert.Equal(new DateTime(1970, 1, 4), prev.ColumnView[0].Values[3]); + Assert.Equal(new DateTime(1970, 1, 6), prev.ColumnView[0].Values[5]); + + // Make sure grain was propagated correctly + Assert.Equal("A", prev.ColumnView[1].Values[1].ToString()); + Assert.Equal("A", prev.ColumnView[1].Values[3].ToString()); + Assert.Equal("A", prev.ColumnView[1].Values[5].ToString()); + + // Make sure forward fill is working as expected. All NA's should be replaced, and imputed rows should have correct values too + Assert.Equal(2.0f, prev.ColumnView[2].Values[1]); + Assert.Equal(2.0f, prev.ColumnView[2].Values[2]); + Assert.Equal(5.0f, prev.ColumnView[2].Values[4]); + Assert.Equal(5.0f, prev.ColumnView[2].Values[5]); + Assert.Equal(5.0f, prev.ColumnView[2].Values[6]); + Assert.Equal(5.0f, prev.ColumnView[2].Values[7]); + + // Make sure IsRowImputed is true for row 1, 3, 5, false for the rest + Assert.Equal(false, prev.ColumnView[3].Values[0]); + Assert.Equal(true, prev.ColumnView[3].Values[1]); + Assert.Equal(false, prev.ColumnView[3].Values[2]); + Assert.Equal(true, prev.ColumnView[3].Values[3]); + Assert.Equal(false, prev.ColumnView[3].Values[4]); + Assert.Equal(true, prev.ColumnView[3].Values[5]); + Assert.Equal(false, prev.ColumnView[3].Values[6]); + Assert.Equal(false, prev.ColumnView[3].Values[7]); + + TestEstimatorCore(pipeline, data); + Done(); + } + [NotCentOS7Fact] public void EntryPoint() { @@ -299,7 +354,50 @@ public void Median() TestEstimatorCore(pipeline, data); Done(); } - + + [NotCentOS7Fact] + public void DateTimeTypeSupportMedian() + { + MLContext mlContext = new MLContext(1); + var dataList = new[] { new { date = new DateTime(1970,1,1), grainA = "A", dataA = 2.0f }, + new { date = new DateTime(1970,1,2), grainA = "A", dataA = float.NaN }, + new { date = new DateTime(1970,1,4), grainA = "A", dataA = 5.0f }}; + var data = mlContext.Data.LoadFromEnumerable(dataList); + + // Build the pipeline, fit, and transform it. + var pipeline = mlContext.Transforms.ReplaceMissingTimeSeriesValues("date", new string[] { "grainA" }, imputeMode: TimeSeriesImputerEstimator.ImputationStrategy.Median, filterColumns: null, suppressTypeErrors: true); + var model = pipeline.Fit(data); + + var output = model.Transform(data); + + var prev = output.Preview(); + + // Should have 3 original columns + 1 more for IsRowImputed + Assert.Equal(4, output.Schema.Count); + + // Imputing one row, so should have length of 4 + Assert.Equal(4, prev.RowView.Length); + + // Check that all rows have the correct dates + Assert.Equal(new DateTime(1970, 1, 1), prev.ColumnView[0].Values[0]); + Assert.Equal(new DateTime(1970, 1, 2), prev.ColumnView[0].Values[1]); + Assert.Equal(new DateTime(1970, 1, 3), prev.ColumnView[0].Values[2]); + Assert.Equal(new DateTime(1970, 1, 4), prev.ColumnView[0].Values[3]); + + // Make sure Median is working as expected. All NA's should be replaced, and imputed rows should have correct values too + Assert.Equal(3.5f, prev.ColumnView[2].Values[1]); + Assert.Equal(3.5f, prev.ColumnView[2].Values[2]); + + // Make sure IsRowImputed is true for imputed row, false for others. + Assert.Equal(false, prev.ColumnView[3].Values[0]); + Assert.Equal(false, prev.ColumnView[3].Values[1]); + Assert.Equal(true, prev.ColumnView[3].Values[2]); + Assert.Equal(false, prev.ColumnView[3].Values[3]); + + TestEstimatorCore(pipeline, data); + Done(); + } + [NotCentOS7Fact] public void Backfill() { From ac67f073c2169ad5fefd930b6249a3c732eba23a Mon Sep 17 00:00:00 2001 From: Michael Sharp Date: Mon, 10 Feb 2020 12:46:57 -0800 Subject: [PATCH 2/3] updates based on PR feedback --- src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs | 15 ++------------- 1 file changed, 2 insertions(+), 13 deletions(-) diff --git a/src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs b/src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs index 00462d649e..2d45af94e0 100644 --- a/src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs +++ b/src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs @@ -61,15 +61,7 @@ public static TimeSeriesImputerEstimator ReplaceMissingTimeSeriesValues(this Tra } /// - /// Imputes missing rows and column data per grain, based on the dates in the date column. This operation needs to happen to every column in the IDataView, - /// If you "filter" a column using the filterColumns and filterMode parameters, if a row is imputed the default value for that type will be used. - /// Currently only float/double/string columns are supported for imputation strategies, and an empty string is considered "missing" for the - /// purpose of this estimator. A new column is added to the schema after this operation is run. The column is called "IsRowImputed" and is a - /// boolean value representing if the row was created as a result of this operation or not. - /// - /// NOTE: It is not recommended to chain this multiple times. If a column is filtered, the default value is placed when a row is imputed, and the - /// default value is not null. Thus any other TimeSeriesImputers will not be able to replace those values anymore causing essentially a very - /// computationally expensive NO-OP. + /// Imputes missing rows and column data per grain, based on the dates in the date column. /// /// /// @@ -93,9 +85,6 @@ public static TimeSeriesImputerEstimator ReplaceMissingTimeSeriesValues(this Tra /// default value is not null. Thus any other TimeSeriesImputers will not be able to replace those values anymore causing essentially a very /// computationally expensive NO-OP. /// - /// The is not a trivial estimator and needs training. - /// - /// /// ]]> /// /// @@ -152,7 +141,7 @@ public enum ImputationStrategy : byte }; /// - /// What the filter strategy used is. + /// Method by which columns are selected for imputing values. /// NoFilter takes all of the columns so you dont have to specify anything. /// Include only does the specified ImputationStrategy on the columns you specify. The other columns will get a default value. /// Exclude is the exact opposite of Include, and does the ImputationStrategy on all columns but the ones you specify, which will get the default value. From 744a3ff62f2667005d292db8e4d3f7b2cd841057 Mon Sep 17 00:00:00 2001 From: Michael Sharp Date: Tue, 25 Feb 2020 13:58:33 -0800 Subject: [PATCH 3/3] Fixes based on PR comments --- .../TimeSeriesImputer.cs | 87 +++++++++++-------- .../TimeSeriesImputerDataView.cs | 78 ++++++++--------- 2 files changed, 87 insertions(+), 78 deletions(-) diff --git a/src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs b/src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs index 2d45af94e0..1c9aeba3c0 100644 --- a/src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs +++ b/src/Microsoft.ML.Featurizers/TimeSeriesImputer.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.IO; using System.Linq; using System.Runtime.InteropServices; using System.Security; @@ -81,6 +82,9 @@ public static TimeSeriesImputerEstimator ReplaceMissingTimeSeriesValues(this Tra /// purpose of this estimator. A new column is added to the schema after this operation is run. The column is called "IsRowImputed" and is a /// boolean value representing if the row was created as a result of this operation or not. /// + /// The imputation strategies that are currently supported are ForwardFill, where the last good value is propagated forward, Backfill, where the next good value is propagated backwards, + /// and Median, where the mathmatical median is used to fill in missing values. + /// /// NOTE: It is not recommended to chain this multiple times. If a column is filtered, the default value is placed when a row is imputed, and the /// default value is not null. Thus any other TimeSeriesImputers will not be able to replace those values anymore causing essentially a very /// computationally expensive NO-OP. @@ -128,14 +132,25 @@ internal sealed class Options : TransformInputBase /// /// This is the representation of which Imputation Strategy to use. - /// ForwardFill takes the value from the last good row and propagates it forward anytime a row is imputer or a missing value is found. - /// BackFill is the same as ForwardFill, expect it takes from the next good row and propagates backwards. + /// ForwardFill takes the value from the last good row and propagates it forward anytime a row is imputed or a missing value is found. + /// BackFill is the same as ForwardFill, except it takes from the next good row and propagates backwards. /// Median only supports float/double, takes the median value found during training and uses that to replace missing values /// public enum ImputationStrategy : byte { + /// + /// Takes the value from the last good row and propagates it forward anytime a row is imputed or a missing value is found. + /// ForwardFill = 1, + + /// + /// Takes the value from the next good row and propagates it backwards anytime a row is imputed or a missing value is found. + /// BackFill = 2, + + /// + /// Takes the median found during training and propagates that anytime a row is imputed or a missing value is found. + /// Median = 3, // Interpolate = 4, interpolate not currently supported in the native code. }; @@ -148,8 +163,19 @@ public enum ImputationStrategy : byte /// public enum FilterMode : byte { + /// + /// Takes all of the columns so you dont have to specify anything. + /// NoFilter = 1, + + /// + /// Only does the specified ImputationStrategy on the columns you specify. The other columns will get a default value. + /// Include = 2, + + /// + /// Does the ImputationStrategy on all columns but the ones you specify, which will get the default value. + /// Exclude = 3 }; @@ -342,7 +368,8 @@ private unsafe TransformerEstimatorSafeHandle CreateTransformerFromEstimator(IDa var allColumns = input.Schema.Where(x => _allColumnNames.Contains(x.Name)).Select(x => TypedColumn.CreateTypedColumn(x, _dataColumns)).ToDictionary(x => x.Column.Name); // Create buffer to hold binary data - var columnBuffer = new byte[4096]; + var memoryStream = new MemoryStream(4096); + var binaryWriter = new BinaryWriter(memoryStream, Encoding.UTF8); // Create TypeId[] for types of grain and data columns; var dataColumnTypes = new TypeId[_dataColumns.Length]; @@ -376,15 +403,17 @@ private unsafe TransformerEstimatorSafeHandle CreateTransformerFromEstimator(IDa while ((fitResult == FitResult.Continue || fitResult == FitResult.ResetAndContinue) && cursor.MoveNext()) { - BuildColumnByteArray(allColumns, ref columnBuffer, out int serializedDataLength); + BuildColumnByteArray(allColumns, ref binaryWriter); - fixed (byte* bufferPointer = columnBuffer) + fixed (byte* bufferPointer = memoryStream.GetBuffer()) { - var binaryArchiveData = new NativeBinaryArchiveData() { Data = bufferPointer, DataSize = new IntPtr(serializedDataLength) }; + var binaryArchiveData = new NativeBinaryArchiveData() { Data = bufferPointer, DataSize = new IntPtr(memoryStream.Position) }; success = FitNative(estimatorHandler, binaryArchiveData, out fitResult, out errorHandle); } if (!success) throw new Exception(GetErrorDetailsAndFreeNativeMemory(errorHandle)); + + memoryStream.Position = 0; } success = CompleteTrainingNative(estimatorHandler, out fitResult, out errorHandle); @@ -401,18 +430,11 @@ private unsafe TransformerEstimatorSafeHandle CreateTransformerFromEstimator(IDa } } - private void BuildColumnByteArray(Dictionary allColumns, ref byte[] columnByteBuffer, out int serializedDataLength) + private void BuildColumnByteArray(Dictionary allColumns, ref BinaryWriter binaryWriter) { - serializedDataLength = 0; foreach (var column in _allColumnNames) { - var bytes = allColumns[column].GetSerializedValue(); - var byteLength = bytes.Length; - if (byteLength + serializedDataLength >= columnByteBuffer.Length) - Array.Resize(ref columnByteBuffer, columnByteBuffer.Length * 2); - - Array.Copy(bytes, 0, columnByteBuffer, serializedDataLength, byteLength); - serializedDataLength += byteLength; + allColumns[column].SerializeValue(ref binaryWriter); } } @@ -542,7 +564,7 @@ internal TypedColumn(DataViewSchema.Column column) } internal abstract void InitializeGetter(DataViewRowCursor cursor); - internal abstract byte[] GetSerializedValue(); + internal abstract void SerializeValue(ref BinaryWriter binaryWriter); internal abstract TypeId GetTypeId(); internal static TypedColumn CreateTypedColumn(DataViewSchema.Column column, string[] optionalColumns) @@ -615,18 +637,14 @@ internal NumericTypedColumn(DataViewSchema.Column column, bool isNullable = fals _isNullable = isNullable; } - internal override byte[] GetSerializedValue() + internal override void SerializeValue(ref BinaryWriter binaryWriter) { dynamic value = GetValue(); - byte[] bytes; - if (value.GetType() == typeof(byte)) - bytes = new byte[1] { value }; - bytes = BitConverter.GetBytes(value); if (_isNullable && value.GetType() != typeof(float) && value.GetType() != typeof(double)) - return new byte[1] { Convert.ToByte(true) }.Concat(bytes).ToArray(); - else - return bytes; + binaryWriter.Write(true); + + binaryWriter.Write(value); } } @@ -640,13 +658,17 @@ internal StringTypedColumn(DataViewSchema.Column column, bool isNullable = false _isNullable = isNullable; } - internal override byte[] GetSerializedValue() + internal override void SerializeValue(ref BinaryWriter binaryWriter) { var value = GetValue().ToString(); var stringBytes = Encoding.UTF8.GetBytes(value); + if (_isNullable) - return new byte[] { Convert.ToByte(true) }.Concat(BitConverter.GetBytes(stringBytes.Length)).Concat(stringBytes).ToArray(); - return BitConverter.GetBytes(stringBytes.Length).Concat(stringBytes).ToArray(); + binaryWriter.Write(true); + + binaryWriter.Write(stringBytes.Length); + + binaryWriter.Write(stringBytes); } } @@ -661,19 +683,16 @@ internal DateTimeTypedColumn(DataViewSchema.Column column, bool isNullable = fal _isNullable = isNullable; } - internal override byte[] GetSerializedValue() + internal override void SerializeValue(ref BinaryWriter binaryWriter) { var dateTime = GetValue(); - byte[] bytes; var value = dateTime.Subtract(_unixEpoch).Ticks / TimeSpan.TicksPerSecond; - bytes = BitConverter.GetBytes(value); - if (_isNullable) - return new byte[1] { Convert.ToByte(true) }.Concat(bytes).ToArray(); - else - return bytes; + binaryWriter.Write(true); + + binaryWriter.Write(value); } } diff --git a/src/Microsoft.ML.Featurizers/TimeSeriesImputerDataView.cs b/src/Microsoft.ML.Featurizers/TimeSeriesImputerDataView.cs index d7baff84ae..3f2ab4d46f 100644 --- a/src/Microsoft.ML.Featurizers/TimeSeriesImputerDataView.cs +++ b/src/Microsoft.ML.Featurizers/TimeSeriesImputerDataView.cs @@ -4,6 +4,7 @@ using System; using System.Collections.Generic; +using System.IO; using System.Linq; using System.Runtime.InteropServices; using System.Security; @@ -24,6 +25,13 @@ internal sealed class TimeSeriesImputerDataView : IDataTransform private TimeSeriesImputerTransformer _parent; public class SharedColumnState { + public SharedColumnState() + { + SourceCanMoveNext = true; + MemStream = new MemoryStream(4096); + BinWriter = new BinaryWriter(MemStream, Encoding.UTF8); + } + public bool SourceCanMoveNext { get; set; } public int TransformedDataPosition { get; set; } @@ -32,7 +40,8 @@ public class SharedColumnState public NativeBinaryArchiveData[] TransformedData { get; set; } // Hold the serialized data that we are going to send to the native code for processing. - public byte[] ColumnBuffer { get; set; } + public MemoryStream MemStream { get; set; } + public BinaryWriter BinWriter { get; set; } public TransformedDataSafeHandle TransformedDataHandler { get; set; } } @@ -54,7 +63,7 @@ internal abstract void InitializeGetter(DataViewRowCursor cursor, TransformerEst string[] grainColumns, string[] dataColumns, string[] allColumnNames, Dictionary allColumns); internal abstract TypeId GetTypeId(); - internal abstract byte[] GetSerializedValue(); + internal abstract void SerializeValue(BinaryWriter binaryWriter); internal abstract unsafe int GetDataSizeInBytes(byte* data, int currentOffset); internal abstract void QueueNonImputedColumnValue(); @@ -152,11 +161,11 @@ internal override unsafe void InitializeGetter(DataViewRowCursor cursor, Transfo NativeBinaryArchiveData* outputData = default; while(outputDataSize == IntPtr.Zero && SharedState.SourceCanMoveNext) { - BuildColumnByteArray(allColumns, allImputedColumnNames, out int bufferLength); + BuildColumnByteArray(allColumns, allImputedColumnNames); QueueDataForNonImputedColumns(allColumns, allImputedColumnNames); - fixed (byte* bufferPointer = SharedState.ColumnBuffer) + fixed (byte* bufferPointer = SharedState.MemStream.GetBuffer()) { - var binaryArchiveData = new NativeBinaryArchiveData() { Data = bufferPointer, DataSize = new IntPtr(bufferLength) }; + var binaryArchiveData = new NativeBinaryArchiveData() { Data = bufferPointer, DataSize = new IntPtr(SharedState.MemStream.Position) }; success = TransformDataNative(transformer, binaryArchiveData, out outputData, out outputDataSize, out errorHandle); if (!success) throw new Exception(GetErrorDetailsAndFreeNativeMemory(errorHandle)); @@ -164,6 +173,8 @@ internal override unsafe void InitializeGetter(DataViewRowCursor cursor, Transfo if (outputDataSize == IntPtr.Zero) SharedState.SourceCanMoveNext = cursor.MoveNext(); + + SharedState.MemStream.Position = 0; } if (!SharedState.SourceCanMoveNext) @@ -238,22 +249,11 @@ internal override void QueueNonImputedColumnValue() SourceQueue.Enqueue(GetSourceValue()); } - private void BuildColumnByteArray(Dictionary allColumns, string[] columns, out int bufferLength) + private void BuildColumnByteArray(Dictionary allColumns, string[] columns) { - bufferLength = 0; foreach(var column in columns.Where(x => x != IsRowImputedColumnName)) { - var bytes = allColumns[column].GetSerializedValue(); - var byteLength = bytes.Length; - if (byteLength + bufferLength >= SharedState.ColumnBuffer.Length) - { - var buffer = SharedState.ColumnBuffer; - Array.Resize(ref buffer, SharedState.ColumnBuffer.Length * 2); - SharedState.ColumnBuffer = buffer; - } - - Array.Copy(bytes, 0, SharedState.ColumnBuffer, bufferLength, byteLength); - bufferLength += byteLength; + allColumns[column].SerializeValue(SharedState.BinWriter); } } @@ -282,21 +282,14 @@ internal NumericTypedColumn(DataViewSchema.Column column, bool isNullable, bool IsNullable = isNullable; } - internal override byte[] GetSerializedValue() + internal override void SerializeValue(BinaryWriter binaryWriter) { dynamic value = GetSourceValue(); - byte[] bytes; - if (value.GetType() == typeof(byte)) - bytes = new byte[1] { value }; - if (BitConverter.IsLittleEndian) - bytes = BitConverter.GetBytes(value); - else - bytes = BitConverter.GetBytes(value); if (IsNullable && value.GetType() != typeof(float) && value.GetType() != typeof(double)) - return new byte[1] { Convert.ToByte(true) }.Concat(bytes).ToArray(); - else - return bytes; + binaryWriter.Write(true); + + binaryWriter.Write(value); } internal override unsafe int GetDataSizeInBytes(byte* data, int currentOffset) @@ -546,13 +539,17 @@ internal StringTypedColumn(DataViewSchema.Column column, bool isNullable, bool i _isNullable = isNullable; } - internal override byte[] GetSerializedValue() + internal override void SerializeValue(BinaryWriter binaryWriter) { var value = GetSourceValue().ToString(); var stringBytes = Encoding.UTF8.GetBytes(value); + if (_isNullable) - return new byte[] { Convert.ToByte(true)}.Concat(BitConverter.GetBytes(stringBytes.Length)).Concat(stringBytes).ToArray(); - return BitConverter.GetBytes(stringBytes.Length).Concat(stringBytes).ToArray(); + binaryWriter.Write(true); + + binaryWriter.Write(stringBytes.Length); + + binaryWriter.Write(stringBytes); } internal unsafe override ReadOnlyMemory GetDataFromNativeBinaryArchiveData(byte* data, int offset) @@ -599,19 +596,16 @@ internal DateTimeTypedColumn(DataViewSchema.Column column, bool isNullable, bool _isNullable = isNullable; } - internal override byte[] GetSerializedValue() + internal override void SerializeValue(BinaryWriter binaryWriter) { var dateTime = GetSourceValue(); - byte[] bytes; var value = dateTime.Subtract(_unixEpoch).Ticks / TimeSpan.TicksPerSecond; - bytes = BitConverter.GetBytes(value); - if (_isNullable) - return new byte[1] { Convert.ToByte(true) }.Concat(bytes).ToArray(); - else - return bytes; + binaryWriter.Write(true); + + binaryWriter.Write(value); } internal unsafe override DateTime GetDataFromNativeBinaryArchiveData(byte* data, int offset) @@ -753,11 +747,7 @@ public Cursor(IChannelProvider provider, DataViewRowCursor input, TransformerEst _schema = schema; _transformer = transformer; - var sharedState = new SharedColumnState() - { - SourceCanMoveNext = true, - ColumnBuffer = new byte[4096] - }; + var sharedState = new SharedColumnState(); _allColumns = _schema.Select(x => TypedColumn.CreateTypedColumn(x, dataColumns, allImputedColumnNames, sharedState)).ToDictionary(x => x.Column.Name); ; _allColumns[IsRowImputedColumnName] = new BoolTypedColumn(_schema[IsRowImputedColumnName], false, true, sharedState);