Быстрый Data Mining или сравнение производительности C# vs Python (pandas-numpy-skilearn)
Всем привет!
Разбираясь со Spark Apache, столкнулся с тем, что после достаточно небольшого усложнения алгоритмов подготовки данных расчеты стали выполняться крайне медленно. Поэтому захотелось реализовать что-нибудь на C# и сравнить производительность с аналогичным по классу решением на стеке python (pandas-numpy-skilearn). Аналогичным, потому что они выполняются на локальной машине. Подготовка данных на C# осуществлялась встроенными средствами (linq), расчет линейной регрессии библиотекой extremeoptimization.
В качестве тестовой использовалась задача «B. Предсказание трат клиентов» с ноябрьского соревнования Sberbank Data Science Journey.
Сразу стоит подчеркнуть, что в данной статье описан исключительно аспект сравнения производительности платформ, а не качества модели и предсказаний.
Итак, сначала краткое описание последовательности действий реализованных на C# (куски кода будут ниже):
1. Загрузить данные из csv. Использовалась библиотека Fast Csv Reader.
2. Отфильтровать расходные операции и выполнить группировку по месяцам.
3. Добавить каждому клиенту те категории, по которым у него не было операций. Для того, чтобы избежать длительный перебор цикл-в-цикле использовал фильтр Блума. Реализацию на C# нашел тут.
4. Формирование массива Hashing trick. Так как готовой реализации под C# не удалось найти, пришлось реализовать самому. Для этого скачал и допилил реализацию хеширования murmurhash3
5. Собственно расчет регрессии.
Решение на Jupyter Notebook (далее JN) выглядит так (подключение библиотек опускаю, потому что это не входило в замеряемое время):
%%time
#Читаем файл с исходными данными по транзакциям
transactions = pd.read_csv('.//JN//SBSJ//transactions.csv')
all_cuses = transactions.customer_id.unique()
#Читаем файл с типами операций
mcc = pd.read_csv('.//JN//SBSJ//tr_mcc_codes.csv', sep=';')
all_mcc = mcc.mcc_code.unique()
#Фильтрация расходных транзакций
transactions = transactions[transactions.amount < 0].copy()
transactions['day'] = transactions.tr_day.apply(lambda dt: dt.split()[0]).astype(int)
transactions.day += 29 - transactions['day'].max()%30
#Преобразование дней в месяцы
transactions['month_num'] = (transactions.day) // 30
train_transactions = transactions[transactions.month_num < 15]
#Добавление отсутствующих типов операций и фильтрация нескольких последних месяцев (на полной выборке не хватало памяти)
grid = list(product(*[all_cuses, all_mcc, range(11, 15)]))
train_grid = pd.DataFrame(grid, columns = ['customer_id', 'mcc_code', 'month_num'])
train = pd.merge(train_grid,
train_transactions.groupby(['month_num', 'customer_id', 'mcc_code'])[['amount']].sum().reset_index(),
how='left').fillna(0)
#Добавление информации о расходах в предыдущих месяцах
for month_shift in range(1, 3):
train_shift = train.copy()
train_shift['month_num'] = train_shift['month_num'] + month_shift
train_shift = train_shift.rename(columns={"amount" : 'amount_{0}'.format(month_shift)})
train_shift = train_shift[['month_num', 'customer_id', 'mcc_code', 'amount_{0}'.format(month_shift)]]
train = pd.merge(train, train_shift, on=['month_num', 'customer_id', 'mcc_code'], how='left').fillna(0)
train['year_num'] = (train.month_num) // 12
#Создание массива hashier trick
hasher = FeatureHasher(n_features=6, input_type='string')
train_sparse = \
hasher.fit_transform(train[['year_num', 'month_num', 'customer_id', 'mcc_code']].astype(str).as_matrix())
train_sparse2 = sparse.hstack([train_sparse, np.log(np.abs(train[['amount_1', 'amount_2']]) + 1).as_matrix(),])
#Без этого хинта расчет регрессии не корректен
d = list(train_sparse2.toarray())
#Собственно расчет
clf = LinearRegression()
clf.fit(d, np.log(np.abs(train['amount']) + 1))
#Результаты
print('Coefficients: \n', clf.coef_)
print('Intercept: \n', clf.intercept_)
print("\nRMSLE: ")
np.sqrt(mse(np.log(np.abs(train['amount']) + 1),clf.predict(d)))
Теперь подробнее о реализации C#. Опыты показали, что классы типа DataTable и прочие очень расточительны по отношению к памяти. Поэтому использовался простой список элементов класса Client:
[Serializable]
public class Client
{
private Int32 name;
private Int16 period;
private Int16 year;
private Int16 mcc;
private double amount;
private double amount1;
private double amount2;
// Методы get/set
...
Далее, чтение данных и группировка:
// Группируем расходные транзакции
List lTransGrouped = lClientsTrans.AsParallel()
.Where(row => row.getAmount() < 0)
.GroupBy(row => new
{
month = (row.getPeriod() + 29 - Convert.ToInt16(maxNumDay) % 30) / 30, // Преобразуем дни в месяцы
mcc = row.getMcc(),
cid = row.getName()
})
.Select(grp => new Client(
grp.Key.cid,
Convert.ToInt16(grp.Key.month),
grp.Key.mcc,
Math.Log(Math.Abs(grp.Sum(r => r.getAmount())) + 1))).ToList();
lClientsTrans = null;
Затем добавляем отсутствующие операции используя фильтр Блума. Можно было бы и без него, но тогда бы увеличилось время выполнения (полный перебор для каждого типа) или объем используемой памяти (если добавлять все типы подряд, а потом агрегировать).
public static List addPeriodMcc(List lTransGrouped, Int16 maxNumMon)
{
List lMcc = new List();
string fnameMcc = @"j:\hadoop\Contest\Contest\tr_mcc_codes.csv";
// Читаем mcc_code
CsvReader csvMccReader = new CsvReader(new StreamReader(fnameMcc), true, ';');
// Читаем типы операций
while (csvMccReader.ReadNextRecord())
{
Int16 mcc = Convert.ToInt16(csvMccReader[0]);
lMcc.Add(new Client(0, 0, mcc, 0));
}
// Готовим таблицу для массива mcc под все записи
List lNewMcc = new List();
// Для генерации отсутствующих записей нужно знать ID клиентов
var lTransCID = lTransGrouped.AsParallel().Select(a => a.getName()).Distinct();
Console.WriteLine("Unique CID: " + lTransCID.Count());
// Задаем мощность фильтра
int capacity = lTransGrouped.Count() * 6; // Чем больше множитель, тем меньше вероятность промахнуться
var filter = new Filter(capacity); //Собственно сам фильтр
// Заполнение фильтра
foreach (var i in lTransGrouped)
filter.Add(i.getName().ToString() + i.getPeriod() + i.getMcc());
// Если записи у клиента нет операции в фильтре, то добавляем
foreach (var cid in lTransCID)
for (Int16 m = 0; m <= maxNumMon; m++)
foreach (var mcc in lMcc)
if (filter.Contains(cid.ToString() + m.ToString() + mcc.getMcc().ToString()) != true)
lNewMcc.Add(new Client(cid, m, mcc.getMcc(), 0));
lTransCID = lMcc = null;
Console.WriteLine("Count lNewMcc: " + lNewMcc.Count);
Console.WriteLine("Count lTransGrouped: " + lTransGrouped.Count);
// Объединение
List lTransFull = lNewMcc.Union(lTransGrouped).ToList();
Console.WriteLine("Count lTransFull: " + lTransFull.Count);
lTransGrouped = lNewMcc = null;
return lTransFull;
}
Этап добавления операций за предыдущие месяцы:
public static List addAmounts(List lTransFull)
{
List lTransFullA2;
// Для корректного добавления значений предыдущих месяцев нужно отсортировать
lTransFullA2 = lTransFull.OrderBy(a => a.getName())
.ThenBy(a => a.getMcc())
.ThenBy(a => a.getYear())
.ThenBy(a => a.getPeriod()).ToList();
int name = 0;
int month = 0;
int year = 0;
int mcc = 0;
int i = 0;
foreach (var l in lTransFullA2)
{
name = l.getName();
mcc = l.getMcc();
year = l.getYear();
month = l.getPeriod();
// Предыдущий месяц
if (i > 0 && name == lTransFullA2[i - 1].getName() &&
mcc == lTransFullA2[i - 1].getMcc() &&
year == lTransFullA2[i - 1].getYear() &&
month == lTransFullA2[i - 1].getPeriod() + 1)
{
l.setAmount1(lTransFullA2[i - 1].getAmount());
}
// Позапрошлый месяц
if (i > 1 && name == lTransFullA2[i - 2].getName() &&
mcc == lTransFullA2[i - 2].getMcc() &&
year == lTransFullA2[i - 2].getYear() &&
month == lTransFullA2[i - 2].getPeriod() + 2)
{
l.setAmount2(lTransFullA2[i - 2].getAmount());
}
i++;
}
return lTransFullA2;
}
Далее заполнение массива hashing trick и подготовка данных в формате понятном модели и собственно расчет:
int n_features = 6;
// Вектор зависимых переменных
Extreme.Mathematics.LinearAlgebra.SparseVector v = Vector.CreateSparse(lTransFullA2.Count);
// Массив независимых (hash + расходы предыдущих месяцев)
md = Matrix.Create(lTransFullA2.Count, n_features + 2);
// Формирование Hashing trick
Parallel.For(0, lTransFullA2.Count(), i => hashing_vectorizer(lTransFullA2[i], i, n_features));
for (int i = 0; i < lTransFullA2.Count; i++)
{
md[i, n_features] = lTransFullA2[i].getAmount1();
md[i, n_features + 1] = lTransFullA2[i].getAmount2();
v.AddAt(i, lTransFullA2[i].getAmount());
}
lTransFullA2 = null;
GC.Collect(2, GCCollectionMode.Forced);
var model = new LinearRegressionModel(v, md); // Формирование модели
model.MaxDegreeOfParallelism = 8;
model.Compute(); // Расчет
Console.WriteLine(model.Summarize()); // Вывод рассчитанных значений
GC.Collect(2, GCCollectionMode.Forced);
Ну и наконец реализация Hashing trick:
public static void hashing_vectorizer(Client f, int i, int n)
{
int[] x = new int[n];
string s = f.getYear().ToString(); //
int idx = getIndx(s, n);
x[idx] += calcBit(s);
md[i,idx] = x[idx];
s = f.getPeriod().ToString();
idx = getIndx(s, n);
x[idx] += calcBit(s);
md[i, idx] = x[idx];
s = f.getName().ToString();
idx = getIndx(s, n);
x[idx] += calcBit(s);
md[i, idx] = x[idx];
s = f.getMcc().ToString();
idx = getIndx(s, n);
x[idx] += calcBit(s);
md[i, idx] = x[idx];
}
// Хэширование для выяснения знака
public static int calcBit(string s)
{
byte b = 0;
b = Convert.ToByte(s[0]);
for (int i = 1; i < s.Count(); i++)
b ^= Convert.ToByte(s[0]);
bool result = true;
while (b >= 1)
{
result ^= (b & 0x01) != 0;
b = Convert.ToByte(b >> 1);
}
if (result)
return -1;
else
return 1;
}
public static int getIndx(string str, int n)
{
Encoding encoding = new UTF8Encoding();
byte[] input = encoding.GetBytes(str);
uint h = MurMurHash3.Hash(input); //
return Convert.ToInt32(h % n);
}
Результаты работы программ практически идентичны (RMSLE около 1.6). Вот как это выглядит:
Теперь переходим к самому интересному - результатам тестирования. Все тесты запускались на i7-2600 (8 потоков, но большую часть времени работало 1-2). Оперативной памяти 12 Гб, ОС Win7.
Для выяснения зависимости времени выполнения от объема данных расчеты запускались на 1.7, 3.4, 5,1 и 6.8 млн. исходных записей (содержимое файла transactions.csv). Но так как в ходе подготовки данных происходила фильтрация за 11-14 месяцы, на графике показано количество данных уже после фильтрации.
Как видно, версия на C# приблизительно в 2 раза быстрее. Похожая ситуация и с расходом памяти. Тут не учитывается память занимаемая Visual Studio (C# запускался в режиме отладки) и браузером (localhost:8888). Для оценки бралось пиковое значение:
При увеличении выборки JN начинал использовать файл подкачки, в результате чего становился бесполезным.
Таким образом, мы видим, что использование C# позволяет существенно быстрее обработать больший объем данных, чем JN, так как оперативная память выступает тут жестким ограничителем.
С другой стороны, средства визуализации matplotlib позволяют анализировать данные почти "на лету", да и кода C# требуется писать гораздо больше. Поэтому в случае нехватки памяти/скорости оптимальным вариантом видится использование стека JN для отладки модели на ограниченной выборке и финальная реализация уже на C#.
Разбираясь со Spark Apache, столкнулся с тем, что после достаточно небольшого усложнения алгоритмов подготовки данных расчеты стали выполняться крайне медленно. Поэтому захотелось реализовать что-нибудь на C# и сравнить производительность с аналогичным по классу решением на стеке python (pandas-numpy-skilearn). Аналогичным, потому что они выполняются на локальной машине. Подготовка данных на C# осуществлялась встроенными средствами (linq), расчет линейной регрессии библиотекой extremeoptimization.
В качестве тестовой использовалась задача «B. Предсказание трат клиентов» с ноябрьского соревнования Sberbank Data Science Journey.
Сразу стоит подчеркнуть, что в данной статье описан исключительно аспект сравнения производительности платформ, а не качества модели и предсказаний.
Итак, сначала краткое описание последовательности действий реализованных на C# (куски кода будут ниже):
1. Загрузить данные из csv. Использовалась библиотека Fast Csv Reader.
2. Отфильтровать расходные операции и выполнить группировку по месяцам.
3. Добавить каждому клиенту те категории, по которым у него не было операций. Для того, чтобы избежать длительный перебор цикл-в-цикле использовал фильтр Блума. Реализацию на C# нашел тут.
4. Формирование массива Hashing trick. Так как готовой реализации под C# не удалось найти, пришлось реализовать самому. Для этого скачал и допилил реализацию хеширования murmurhash3
5. Собственно расчет регрессии.
Решение на Jupyter Notebook (далее JN) выглядит так (подключение библиотек опускаю, потому что это не входило в замеряемое время):
%%time
#Читаем файл с исходными данными по транзакциям
transactions = pd.read_csv('.//JN//SBSJ//transactions.csv')
all_cuses = transactions.customer_id.unique()
#Читаем файл с типами операций
mcc = pd.read_csv('.//JN//SBSJ//tr_mcc_codes.csv', sep=';')
all_mcc = mcc.mcc_code.unique()
#Фильтрация расходных транзакций
transactions = transactions[transactions.amount < 0].copy()
transactions['day'] = transactions.tr_day.apply(lambda dt: dt.split()[0]).astype(int)
transactions.day += 29 - transactions['day'].max()%30
#Преобразование дней в месяцы
transactions['month_num'] = (transactions.day) // 30
train_transactions = transactions[transactions.month_num < 15]
#Добавление отсутствующих типов операций и фильтрация нескольких последних месяцев (на полной выборке не хватало памяти)
grid = list(product(*[all_cuses, all_mcc, range(11, 15)]))
train_grid = pd.DataFrame(grid, columns = ['customer_id', 'mcc_code', 'month_num'])
train = pd.merge(train_grid,
train_transactions.groupby(['month_num', 'customer_id', 'mcc_code'])[['amount']].sum().reset_index(),
how='left').fillna(0)
#Добавление информации о расходах в предыдущих месяцах
for month_shift in range(1, 3):
train_shift = train.copy()
train_shift['month_num'] = train_shift['month_num'] + month_shift
train_shift = train_shift.rename(columns={"amount" : 'amount_{0}'.format(month_shift)})
train_shift = train_shift[['month_num', 'customer_id', 'mcc_code', 'amount_{0}'.format(month_shift)]]
train = pd.merge(train, train_shift, on=['month_num', 'customer_id', 'mcc_code'], how='left').fillna(0)
train['year_num'] = (train.month_num) // 12
#Создание массива hashier trick
hasher = FeatureHasher(n_features=6, input_type='string')
train_sparse = \
hasher.fit_transform(train[['year_num', 'month_num', 'customer_id', 'mcc_code']].astype(str).as_matrix())
train_sparse2 = sparse.hstack([train_sparse, np.log(np.abs(train[['amount_1', 'amount_2']]) + 1).as_matrix(),])
#Без этого хинта расчет регрессии не корректен
d = list(train_sparse2.toarray())
#Собственно расчет
clf = LinearRegression()
clf.fit(d, np.log(np.abs(train['amount']) + 1))
#Результаты
print('Coefficients: \n', clf.coef_)
print('Intercept: \n', clf.intercept_)
print("\nRMSLE: ")
np.sqrt(mse(np.log(np.abs(train['amount']) + 1),clf.predict(d)))
Теперь подробнее о реализации C#. Опыты показали, что классы типа DataTable и прочие очень расточительны по отношению к памяти. Поэтому использовался простой список элементов класса Client:
[Serializable]
public class Client
{
private Int32 name;
private Int16 period;
private Int16 year;
private Int16 mcc;
private double amount;
private double amount1;
private double amount2;
// Методы get/set
...
Далее, чтение данных и группировка:
// Группируем расходные транзакции
List lTransGrouped = lClientsTrans.AsParallel()
.Where(row => row.getAmount() < 0)
.GroupBy(row => new
{
month = (row.getPeriod() + 29 - Convert.ToInt16(maxNumDay) % 30) / 30, // Преобразуем дни в месяцы
mcc = row.getMcc(),
cid = row.getName()
})
.Select(grp => new Client(
grp.Key.cid,
Convert.ToInt16(grp.Key.month),
grp.Key.mcc,
Math.Log(Math.Abs(grp.Sum(r => r.getAmount())) + 1))).ToList();
lClientsTrans = null;
Затем добавляем отсутствующие операции используя фильтр Блума. Можно было бы и без него, но тогда бы увеличилось время выполнения (полный перебор для каждого типа) или объем используемой памяти (если добавлять все типы подряд, а потом агрегировать).
public static List addPeriodMcc(List lTransGrouped, Int16 maxNumMon)
{
List lMcc = new List();
string fnameMcc = @"j:\hadoop\Contest\Contest\tr_mcc_codes.csv";
// Читаем mcc_code
CsvReader csvMccReader = new CsvReader(new StreamReader(fnameMcc), true, ';');
// Читаем типы операций
while (csvMccReader.ReadNextRecord())
{
Int16 mcc = Convert.ToInt16(csvMccReader[0]);
lMcc.Add(new Client(0, 0, mcc, 0));
}
// Готовим таблицу для массива mcc под все записи
List lNewMcc = new List();
// Для генерации отсутствующих записей нужно знать ID клиентов
var lTransCID = lTransGrouped.AsParallel().Select(a => a.getName()).Distinct();
Console.WriteLine("Unique CID: " + lTransCID.Count());
// Задаем мощность фильтра
int capacity = lTransGrouped.Count() * 6; // Чем больше множитель, тем меньше вероятность промахнуться
var filter = new Filter(capacity); //Собственно сам фильтр
// Заполнение фильтра
foreach (var i in lTransGrouped)
filter.Add(i.getName().ToString() + i.getPeriod() + i.getMcc());
// Если записи у клиента нет операции в фильтре, то добавляем
foreach (var cid in lTransCID)
for (Int16 m = 0; m <= maxNumMon; m++)
foreach (var mcc in lMcc)
if (filter.Contains(cid.ToString() + m.ToString() + mcc.getMcc().ToString()) != true)
lNewMcc.Add(new Client(cid, m, mcc.getMcc(), 0));
lTransCID = lMcc = null;
Console.WriteLine("Count lNewMcc: " + lNewMcc.Count);
Console.WriteLine("Count lTransGrouped: " + lTransGrouped.Count);
// Объединение
List lTransFull = lNewMcc.Union(lTransGrouped).ToList();
Console.WriteLine("Count lTransFull: " + lTransFull.Count);
lTransGrouped = lNewMcc = null;
return lTransFull;
}
Этап добавления операций за предыдущие месяцы:
public static List addAmounts(List lTransFull)
{
List lTransFullA2;
// Для корректного добавления значений предыдущих месяцев нужно отсортировать
lTransFullA2 = lTransFull.OrderBy(a => a.getName())
.ThenBy(a => a.getMcc())
.ThenBy(a => a.getYear())
.ThenBy(a => a.getPeriod()).ToList();
int name = 0;
int month = 0;
int year = 0;
int mcc = 0;
int i = 0;
foreach (var l in lTransFullA2)
{
name = l.getName();
mcc = l.getMcc();
year = l.getYear();
month = l.getPeriod();
// Предыдущий месяц
if (i > 0 && name == lTransFullA2[i - 1].getName() &&
mcc == lTransFullA2[i - 1].getMcc() &&
year == lTransFullA2[i - 1].getYear() &&
month == lTransFullA2[i - 1].getPeriod() + 1)
{
l.setAmount1(lTransFullA2[i - 1].getAmount());
}
// Позапрошлый месяц
if (i > 1 && name == lTransFullA2[i - 2].getName() &&
mcc == lTransFullA2[i - 2].getMcc() &&
year == lTransFullA2[i - 2].getYear() &&
month == lTransFullA2[i - 2].getPeriod() + 2)
{
l.setAmount2(lTransFullA2[i - 2].getAmount());
}
i++;
}
return lTransFullA2;
}
Далее заполнение массива hashing trick и подготовка данных в формате понятном модели и собственно расчет:
int n_features = 6;
// Вектор зависимых переменных
Extreme.Mathematics.LinearAlgebra.SparseVector v = Vector.CreateSparse(lTransFullA2.Count);
// Массив независимых (hash + расходы предыдущих месяцев)
md = Matrix.Create(lTransFullA2.Count, n_features + 2);
// Формирование Hashing trick
Parallel.For(0, lTransFullA2.Count(), i => hashing_vectorizer(lTransFullA2[i], i, n_features));
for (int i = 0; i < lTransFullA2.Count; i++)
{
md[i, n_features] = lTransFullA2[i].getAmount1();
md[i, n_features + 1] = lTransFullA2[i].getAmount2();
v.AddAt(i, lTransFullA2[i].getAmount());
}
lTransFullA2 = null;
GC.Collect(2, GCCollectionMode.Forced);
var model = new LinearRegressionModel(v, md); // Формирование модели
model.MaxDegreeOfParallelism = 8;
model.Compute(); // Расчет
Console.WriteLine(model.Summarize()); // Вывод рассчитанных значений
GC.Collect(2, GCCollectionMode.Forced);
Ну и наконец реализация Hashing trick:
public static void hashing_vectorizer(Client f, int i, int n)
{
int[] x = new int[n];
string s = f.getYear().ToString(); //
int idx = getIndx(s, n);
x[idx] += calcBit(s);
md[i,idx] = x[idx];
s = f.getPeriod().ToString();
idx = getIndx(s, n);
x[idx] += calcBit(s);
md[i, idx] = x[idx];
s = f.getName().ToString();
idx = getIndx(s, n);
x[idx] += calcBit(s);
md[i, idx] = x[idx];
s = f.getMcc().ToString();
idx = getIndx(s, n);
x[idx] += calcBit(s);
md[i, idx] = x[idx];
}
// Хэширование для выяснения знака
public static int calcBit(string s)
{
byte b = 0;
b = Convert.ToByte(s[0]);
for (int i = 1; i < s.Count(); i++)
b ^= Convert.ToByte(s[0]);
bool result = true;
while (b >= 1)
{
result ^= (b & 0x01) != 0;
b = Convert.ToByte(b >> 1);
}
if (result)
return -1;
else
return 1;
}
public static int getIndx(string str, int n)
{
Encoding encoding = new UTF8Encoding();
byte[] input = encoding.GetBytes(str);
uint h = MurMurHash3.Hash(input); //
return Convert.ToInt32(h % n);
}
Результаты работы программ практически идентичны (RMSLE около 1.6). Вот как это выглядит:
Теперь переходим к самому интересному - результатам тестирования. Все тесты запускались на i7-2600 (8 потоков, но большую часть времени работало 1-2). Оперативной памяти 12 Гб, ОС Win7.
Для выяснения зависимости времени выполнения от объема данных расчеты запускались на 1.7, 3.4, 5,1 и 6.8 млн. исходных записей (содержимое файла transactions.csv). Но так как в ходе подготовки данных происходила фильтрация за 11-14 месяцы, на графике показано количество данных уже после фильтрации.
Как видно, версия на C# приблизительно в 2 раза быстрее. Похожая ситуация и с расходом памяти. Тут не учитывается память занимаемая Visual Studio (C# запускался в режиме отладки) и браузером (localhost:8888). Для оценки бралось пиковое значение:
При увеличении выборки JN начинал использовать файл подкачки, в результате чего становился бесполезным.
Таким образом, мы видим, что использование C# позволяет существенно быстрее обработать больший объем данных, чем JN, так как оперативная память выступает тут жестким ограничителем.
С другой стороны, средства визуализации matplotlib позволяют анализировать данные почти "на лету", да и кода C# требуется писать гораздо больше. Поэтому в случае нехватки памяти/скорости оптимальным вариантом видится использование стека JN для отладки модели на ограниченной выборке и финальная реализация уже на C#.