Как использовать Microsoft SQL для отчётов в Power BI. На примере Mindbox

Если вы хотите понимать, насколько эффективны вложения в рекламу, нужно всё измерять: письма, визиты, заказы, выручку. Важна и скорость получения этих метрик. Для наглядности можно строить красивые и понятные отчёты в Microsoft Power BI. А упростить получение данных для отчётов поможет их хранение в собственной базе. Это удобно, особенно когда информации много, и получать её напрямую из онлайн-сервисов затруднительно. Мы храним данные в Microsoft SQL, но можно использовать и другой сервер БД.

Проблемы построения отчётов по данным из ESP

Современные почтовые платформы (Email Sending Platform или ESP) собирают статистику по рассылкам: отправки, доставки, открытия, клики. А более продвинутые платформы, например, Mindbox, позволяют добавить сюда заказы и выручку.

Кроме того, данные можно и нужно собирать при помощи Google Analytics и/или Яндекс.Метрики.

Данные где-то хранятся, доступны через API почтовых платформ, но строить отчёты по этим сырым данным сложно, поскольку:

1. Данные могут храниться в разных форматах, зачастую неудобных.
2. Доступ к данным может быть затруднительным из-за сложных схем авторизации, сложного протокола получения данных.
3. Данные могут храниться в ESP только за последние несколько месяцев, более старые данные платформы обычно удаляют.
4. Когда данных много, то оперативно получить данные невозможно из-за ограничений в производительности API почтовых платформ или пропускной способности сети.
5. При организации автоматического обновления данных в отчёте на сервере Power BI часто возникают проблемы с доступом к источникам из-за политики конфиденциальности.

Для построения одного отчёта может потребоваться сотни или даже тысячи API-запросов. Например, чтобы построить отчёт по сегментам подписчиков, нужно сделать столько запросов, сколько выделено сегментов аудитории. Если нужно анализировать рост сегментов по дням, то нужно выполнить эти запросы столько раз, за сколько дней мы хотим построить отчёт. Даже если один запрос выполняется секунду, в Power BI могут возникнуть проблемы при первоначальной загрузке данных и обновлении.

Поэтому мы стали предварительно собирать данные в систему управления базами данных — Microsoft SQL Server. Можно собирать данные и в другие базы, в зависимости от наших задач. При отсутствии своего SQL-сервера часто используют облачные решения, такие как MS Azure или BigQuery.

Кейс от аналитиков: клиенту потребовалась статистика по ежедневной динамике сегментов подписчиков. Используемая ESP-система пересчитывает сегменты и отдаёт статистику только два раза в месяц. С помощью ежедневного агрегирования данных на нашем SQL-сервере мы смогли удовлетворить запрос клиента. Конечно, исторических данных нет, информация собирается с момента поступления заявки от клиента, но… уже собраны данные за 3 года!

Результат стандартного запроса к API ESP для получения размера сегмента выглядит так:

Как видим, довольно скромная информация: размер сегмента и дата пересчёта.

При агрегировании данных на SQL-сервере можем собрать информацию за необходимый период.

Выбранный для примера сегмент оказался одним из главных и пересчитывался довольно часто, поэтому с SQL-сервера обычным запросом можно получить почти ежедневную статистику.

Преимущества и особенности работы с Microsoft SQL Server

В SQL-базе хранятся уже подготовленные данные, удобные для построения отчёта. Для наполнения этой базы мы пишем серверные скрипты на программной платформе Node.js. Она имеет весь необходимый инструментарий и позволяет эффективно распараллеливать API-запросы.

Скрипты при помощи API-запросов получают данные из ESP, преобразуют в удобный формат и сохраняют в таблицах БД. Мы настроили планировщик задач так, чтобы эти скрипты выполнялись периодически по ночам.

Таким образом, у нас в базе актуальные данные, проблем с загрузкой и обновлением данных в отчёты Power BI не возникает: данные хранятся столько, сколько мы захотим.

Например, мы импортируем данные о действиях по сегментам из Mindbox. В данном случае действие — это событие, которое происходило с письмом, отправленным по сегменту «отправлено→доставлено→открыто→кликнули по ссылке в письме→попало в спам».

При написании подобных скриптов надо обратить внимание на следующие не очевидные, но важные моменты.

1. Mindbox отдаёт данные не единым потоком, а постранично. Максимальное количество записей на один запрос — 10 000, но лучше забирать по 500-1000 записей — так меньше ошибок API. Поэтому на один сегмент придётся сделать несколько запросов. Важно обеспечить последовательность этих запросов, чтобы данные хранились в том же порядке, как у Mindbox.
2. В API Mindbox есть возможность получать данные с фильтрацией. Нам важно фильтровать c учётом ID (уникального ключа) действия, а значит, получать только те действия, ID которых больше начального, если мы собираем данные за определённый период.
3. Скрипт может выполняться очень долго, поэтому важно сделать механизм пополнения собранных данных. Тогда можно ежедневно дописывать данные, а не собирать всё сначала.
4. Любой API-запрос к Mindbox может закончиться ошибкой по разным причинам. Скрипт должен корректно обрабатывать ошибки, чтобы собранные данные были корректными и полными, без пропусков. Если возникает ошибка, то мы завершаем скрипт сбора данных. Данные будут дособраны при следующем запуске скрипта.
5. Mindbox отдаёт «сырые» данные, которые чаще всего не нужны для построения отчётности. На SQL-сервере можно сразу провести агрегацию данных, что увеличит скорость загрузки и упростит их обработку в отчётах Power BI.

По этому поводу мы получили комментарий менеджера продуктов Mindbox Ильи Цырульникова:

«В июне 2020 года мы выпускаем новый API экспортов, который работает асинхронно и возвращает сразу большой объём данных. Например, можно будет одним запросом выгрузить все заказы за последние полгода.

В будущем мы планируем сделать стандартный провайдер для одной из BI-систем. Сырые данные в BI будут поступать автоматически без развертывания и поддержки отдельной базы данных».

Как работает сборщик данных Microsoft SQL

Алгоритм программы можно описать примерно так:

1. Узнать максимальный ID действия, который мы успели сохранить в нашей базе.
2. Получить через API Mindbox действия, ID которых больше, чем максимальный ID в нашей базе. Если данных пока нет, то начинать с нулевого ID.
3. Если данные получены, то записываем их в нашу базу данных. Возвращаемся к пункту 1. Если же данных нет, значит, мы либо собрали их все раньше, либо Mindbox вернул ошибку.

Повторный запуск программы будет дописывать данные. Так мы можем ежедневно пополнять базу данных и актуализировать её.

/* MSSQL */
const config = require('./config.js');
/**
Ожидаемое содержимое конфигурационного файла:
module.exports = {

mssql: {
user: 'sa',
password: 'mssqlpassword',
server: 'mssqlhost',
database: 'mssqldbname',
connectionTimeout: 60000,
requestTimeout: 60000
},

mindbox_api_key: 'ключ_доступа_к_API_mindbox'
};
**/

const async = require('async');
const request = require('request');
const https = require("https");
const sql = require('mssql');
var parseString = require('xml2js').parseString;
var _ = require('lodash');
var iconv = require('iconv-lite');

//Подключение к SQL-серверу
sql.connect(config.mssql, err => {
// Query
if (err) {
console.log('SQL connect error: ', err, config.mssql);
}
})

sql.on('error', err => {
console.log('SQL server error: ', err);
// ... error handler
})

setTimeout(function() {
//Будем испортировать данные по нескольким операциям, сохраняя данные в одноименные таблицы
var operarions = ['EksportDejstvijPoRuchnym', 'EksportDejstvijPoRassylkam', 'EksportDejstvijPoRassylkamSkandinavskij', 'EksportDejstvijPoRassylkamNovoe', 'EksportDejstvijPoRassylkamSkolkovskij', 'EksportDejstvijPoRassylkamDsk1', 'EksportDejstvijPoRassylkamDyxanie', 'EksportDejstvijPoRassylkamPokolenie', 'EksportDejstvijPoRassylkamFsk'];
async.forEach(operarions, function(operation, cb1) {
mbxChunk(operation, cb1);
},
function(err, res) {
if (err) {
sql.close();
process.exit();
return;
}
console.log('DONE!');
sql.close();
process.exit();
}
)
}, 1000);

/**
* Рекурсивная функция получения данных и записи в таблицу
*/
function mbxChunk(operation, cb) {
console.log('New Chunk '+operation);
var sqlreq = new sql.Request();
//Получаем максимальный mindboxId из таблицы, чтобы знать, с какого ID собирать данные
sqlreq.query("SELECT MAX(mindboxId) AS last FROM "+config.mssql.database+".dbo."+operation+";", function(err, maxres) {
if (err || !maxres.recordset[0]) {
console.log('MsSQL SELECT MAX error: ', err);
cb(err);
return;
}
var chunk = parseInt(maxres.recordset[0].last || 0) + 1;

//Получим данные, начиная с ID=chunk
mbxCall(operation, chunk, function(err, xml) {
console.log(operation + ' from #' + chunk.toString() + ' started');
if (err) {
console.log(operation + ' from #' + chunk.toString() + ': Mindbox API request error: ', err);
cb(err);
}
parseString(xml, function(err, res) {
if (err) {
console.log(operation + ' from #' + chunk.toString() + ': Mindbox returns invalid XML, parse error: ', err);
console.log(xml);
cb(err);
}
if (res && res.result && res.result.status == 'Success') {
var rows = res.result.customerActions ? res.result.customerActions[0].customerAction : [];
if (rows.length > 0) {
mbxHandle(operation, rows, function(err, res) {
console.log(operation + ' from #' + chunk.toString() + ' + '+rows.length.toString()+' done!');
mbxChunkCopy(operation, cb);
});
}
else {
console.log(operation + ' - data is over!');
cb();
}
}
else {
console.log(operation + ' from #' + chunk.toString() + ': Mindbox returns unsuccessfull result: ', res);
cb(err);
}
});
})
})
}

function mbxChunkCopy(operation, cb) {
mbxChunk(operation, cb);
}

/**
* Запрос к API Mindbox
* Получаем 1000 записей. ID записи начинается со startId
*/
function mbxCall(operation, startId, cb) {
var request = require("request");
var options = {
method: 'POST',
url: 'https://api.mindbox.ru/v3/operations/sync',
qs: {
endpointId: 'fskuniversalpoint', operation: operation
},
headers: {
"Cache-Control": "no-cache",
"Content-Type": "application/xml",
"Accept": "application/xml",
"Authorization": 'Mindbox secretKey="'+config.mindbox_api_key+'"'
},
body: '<operation>\r\n  <page>\r\n    <firstmindboxid>' + startId.toString() + '</firstmindboxid>\r\n    <pagenumber>1</pagenumber>\r\n    <itemsperpage>1000</itemsperpage>\r\n  </page>\r\n</operation>'
};

request(options, function (error, response, body) {
cb(error, body);
});
}

/**
* Обработка блока данных от Mindbox
* Записать в таблицу table MsSQL
*/
function mbxHandle(table, rows, cb) {
console.log(rows.length.toString() + ' items to save into '+table);
var i = 0;

async.forEach(rows, function(row, cb1) {
try {
var sqlreq = new sql.Request();
sqlreq.input('mindboxId', sql.Int, parseInt(row.ids[0].mindboxId[0]));
sqlreq.input('transactionId', sql.VarChar, (row.ids[0].transactionId ? row.ids[0].transactionId[0] : ''));
sqlreq.input('systemName', row.actionTemplate[0].systemName[0]);
sqlreq.input('name', sql.NVarChar, row.actionTemplate[0].name[0]);
sqlreq.input('dateTimeUtc', sql.DateTime, new Date(row.dateTimeUtc[0]));
sqlreq.input('pointOfContact_externalId', sql.VarChar, row.pointOfContact[0].ids[0].externalId[0]);
sqlreq.input('customer_mindboxId', sql.VarChar, (row.customer[0].ids[0] ? row.customer[0].ids[0].mindboxId[0] : ''));
sqlreq.input('customer_webSiteId', sql.VarChar, (row.customer[0].ids[0].webSiteId ? row.customer[0].ids[0].webSiteId[0] : ''));
sqlreq.input('channel', sql.VarChar, (row.mailing && row.mailing[0].channel ? row.mailing[0].channel[0] : ''));
sqlreq.input('action', sql.VarChar, (row.mailing && row.mailing[0].action ? row.mailing[0].action[0] : ''));
sqlreq.input('notSentReason_name', sql.VarChar, row.ids[0].mindboxId[0]);
sqlreq.input('notSentReason_systemName', sql.VarChar, (row.mailing && row.mailing[0].notSentReason &&  row.mailing[0].notSentReason[0].name ? row.mailing[0].notSentReason[0].name[0] : ''));
sqlreq.input('groupingKey', sql.VarChar, (row.mailing && row.mailing[0].groupingKey ? row.mailing[0].groupingKey[0] : ''));
sqlreq.input('link', sql.VarChar, (row.mailing && row.mailing[0].link ? row.mailing[0].link[0] : ''));
sqlreq.input('viewEmailMessageUrl', sql.VarChar, (row.mailing && row.mailing[0].viewEmailMessageUrl ? row.mailing[0].viewEmailMessageUrl[0] : ''));
var q = "INSERT INTO "+config.mssql.database+".dbo."+table
+ " (mindboxId, transactionId, systemName, name, dateTimeUtc, pointOfContact_externalId, customer_mindboxId, customer_webSiteId, channel, [action], notSentReason_name, notSentReason_systemName, groupingKey, link, viewEmailMessageUrl)"
+ " VALUES(@mindboxId, @transactionId, @systemName, @name, @dateTimeUtc, @pointOfContact_externalId, @customer_mindboxId, @customer_webSiteId, @channel, @action, @notSentReason_name, @notSentReason_systemName, @groupingKey, @link, @viewEmailMessageUrl);";
}
catch (err) {
console.log('MsSQL input vars error: ', err, JSON.stringify(row));
return cb1(err);
}
sqlreq.query(q, function(err, r) {
if (err) {
console.log('MsSQL query error: ', err, JSON.stringify(row));
}
cb1(err, r);
})
}, function(err, r) {
cb(err, r);
});
}

Когда данные собраны, мы по ним можем построить, например, вот такие отчёты.

В редакторе запросов Power BI можно организовать прямую загрузку данных из Mindbox с помощью такого запроса:

let
// функция запрашивает очередную страницу данных с Mindbox
   ExportAction = (name as text, page as number)=>let
   Source=Xml.Tables(Web.Contents("https://api.mindbox.ru/v3/",
                     [Headers=[#"content-type"="application/xml",Accept="application/xml",Authorization="Mindbox secretKey=""ХХХХХХХХХХХХХХХХХХХХХХХХ"""],
                     RelativePath="operations/sync",
                     Query=[endpointId="ХХХХХХХХХХХХХХХХХ",operation=name],
                     Content=Text.ToBinary("<operation><page><firstmindboxid>0</firstmindboxid><pagenumber>"&Number.ToText(page)&"</pagenumber><itemsperpage>10000</itemsperpage></page></operation>")])),
   Data= try Source{0}[customerActions] otherwise null // проверяем что ответ непустой
in
   Data,
// начало программы
   name="XXXXXXXXXXX",  // задаём имя действия в Mindbox
   Source = List.Generate( ()=>                                            // циклом запрашиваем все страницы с данными от Mindbox
            [Result = ExportAction(name, 1), Page = 1],                    // запрос первой страницы данных.
            each [Result] <> null,                                         // если ответ пустой, то конец цикла
            each [Result = ExportAction(name,[Page]+1), Page = [Page]+1],  // иначе запрашиваем следующую страницу данных
            each [Result]),
// конвертируем полученные данные в таблицу
    #"Converted to Table" = Table.FromList(Source, Splitter.SplitByNothing(), null, null, ExtraValues.Error)
//   *
//   *
//   обработка полученных данных
//   *
//   *
//   result = ..........
in
    result

И он даже будет работать. Часа 2-3. И, возможно, данные вы соберёте, если у вас не так много подписчиков и рассылки выходят не очень часто.

В противном случае вы, скорее всего, получите сообщение, что сервер временно недоступен. И данные загружены не будут. При возникновении ошибки запрос прекратит свою работу, а все загруженные данные не сохранятся.

После апдейта в феврале 2020 в PowerQuery появилась возможность агрегировать (догружать) данные, но пока этот способ работает с множеством ограничений, да и не каждый API позволит реализовать заложенный алгоритм.

Если в Power BI загружать данные, заранее собранные на SQL-сервере, то запрос будет гораздо проще и выполняться будет гораздо быстрее — всего несколько секунд.

А выглядеть будет так (сравните с запросом к API):

let
Source = Sql.Database("ServerName«, «DataBase»),
Result = Source{[Schema="dbo",Item="TableName"]}[Data]
in
Result

А если вы заранее продумали и правильно сформировали структуру таблиц на сервере, дополнительное преобразование и обработка данных в редакторе запросов Power query не потребуются.

Как выглядит отчёт в Power BI

Вот такой отчёт можно собрать на данных из Mindbox и Google Analytics.

Следуя изложенной выше методике, каждую ночь в базу данных на SQL-сервере собираются данные из Mindbox и Google Analytics (специальный скрипт собирает данные без сэмплирования), затем собранные и агрегированные данные с SQL-сервера загружаются в отчёт на сервере Power BI.

Таким образом, пользователи уже к началу рабочего дня получают в отчёте актуальные данные.