[Перевод] Строим ETL-конвейер для машинного обучения с помощью Kafka, Clickhouse и Go03.10.2024 18:00

Эта программа немного сложнее предыдущих. Вот краткий обзор того, что здесь происходит:

• В горутинах readWithReader () используются несколько читателей Kafka одной и той же группы потребителей — они передают необработанные сообщения по каналу msgs в виде массива байтов. В то же время горутины transform () считывают из этого канала каждое сообщение, чтобы валидировать и преобразовать его в структуру Entry для передачи через канал entries в конечный набор горутин, batchInsert (), где они будут внесены в Clickhouse.

• В программе также запущена горутина, которая ловит сигнал ctrl-c (SIGINT) и закрывает соединение читателей и Clickhouse для изящного завершения работы. После создания всех горутин, main остановится в ожидании, пока этот сигнал пройдет по каналу signal_ch, прежде чем приступить к очистке.

//(скопировано из main)
go func (signal_ch chan os.Signal){
       signal.Notify(signal_ch, os.Interrupt) // ловим нажатие клавиши Ctrl-C

   } (signal_ch)

//...

<-signal_ch // ожидает до тех пор, пока соединения не будут готовы к очистке

   for i,reader := range readers{
       fmt.Printf("Closing reader: %d\n",i)
       if err := reader.Close(); err != nil {
           fmt.Println("failed to close reader:", err)
       }
   }
// остальная часть очистки...

Следует отметить несколько моментов:

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

kafka-consumer-groups.sh --describe --bootstrap-server localhost:9092 --all-groups

в shell запущенного контейнера kafka, чтобы увидеть смещения сообщений, потребляемых из топика.

• Валидация на этапе преобразования заключается исключительно в проверке того, что нужные нам поля ненулевые (отсутствующие поля при преобразовании инициализируются умолчательными значениями). В зависимости от данных, некоторые записи могут быть отброшены ошибочно (например, если в исходных данных есть температура 0 градусов).

//Разбиваем сообщения из канала []byte на Entries и проверяем их
//перед отправкой их по каналу Entry
func transform(ch chan []byte, routine_num int, entries chan Entry){
   var count int = 0
   for val := range ch{
       var temp Entry
       err := json.Unmarshal(val, &temp) //в структуру
       if err != nil{
           //отмечаем, но двигаемся дальше
           fmt.Println("Error unmarshaling: " + string(val))
           continue
       }
       if validify(&temp) != nil{
           fmt.Printf("Error transforming data: %v\n", temp)
           continue
       }
       entries <- temp
       count++
       fmt.Printf("TRANSFORMED(%d): %v\n", count, routine_num)
   }
}

//Проверяем, что json-строка, преобразованная в структуру, содержит корректные данные (не нули)
func validify(entry *Entry) (error){
   //преобразование даты и времени в то, что clickhouse может преобразовать в DateTime
   res, err := time.Parse("01/02/06-03:04PM", entry.Date_time)
   if err != nil{
       return err
   }
   //если какая-либо запись равна 0 (отсутствующие данные), отбрасываем ее

   // if(entry.Sea_level_pressure==0||entry.Altimeter==0||
   //  entry.Air_temp==0||entry.Relative_humidity==0||entry.Dew_point_temperature==0||
   //  entry.Wind_direction==0||entry.Wind_speed==0){
       if(entry.Air_temp==0){
           return errors.New("invalid entry")
       }
   entry.Date_time = res.Format(time.DateTime)
   return nil
}
// продолжение ниже...

• Для парсинга нестандартных форматов дат в строках нам пригодится Parse () из пакета time. Вот так будет выглядеть строка, представляющая 3:04 PM 2 января 2006 года в нужном для парсинга формате (например,»01/02/06–03:04PM») (https://pkg.go.dev/time#Parse).

//скопировано
res, err := time.Parse("01/02/06-03:04PM", entry.Date_time)

• Функция AppendStruct () из API Clickhouse-go поможет нам с разбиением нашей структуры Entry на строки и добавлением их в батч (в качестве альтернативы вызову Append () и явной передачи каждого столбца).

// Вставка батчей в таблицу из канала Entry
func batchInsert(conn clickhouse.Conn, ch chan Entry, table string, routine_num int){
 var count int = 0
 for val := range ch{
       batch, _ := conn.PrepareBatch(context.Background(), "INSERT INTO " + table)
       batch.AppendStruct(&val) // ПОЛЯ СТРУКТУРЫ ДОЛЖНЫ НАЧИНАТЬСЯ С ВЕРХНЕГО РЕГИСТРА, иначе это не будет работать
       count++;
       batch.Send()
       fmt.Printf("INSERTED(%d): %d\n", count, routine_num)
 }
}
// конец программы

Машинное обучение в Clickhouse

predict-set-up.sh (ML)

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

Вот как выглядит таблица с нашими данными (это лишь часть таблицы):

А вот часть того, что мы хотим (и будем) скармливать нашей модели (таблица model_data):

В массиве temp хранится окно с предшествующими температурами (30 штук) для каждой даты, затем температура конкретной даты, а в массиве time — смещение времени от начала окна для каждой из них. Например, дата 8/8/24 3:53 (первая строка в таблице) имеет в своем массиве temp температуры дат 8/7/24 2:53, 8/7/24 3:53, 8/7/24 4:53 и т. д. Поскольку начало этого диапазона приходится на 8/7/24 2:53, смещение по времени для каждой записи начинается с 0, затем растет на 60 минут (3600) с 2:53 → 3:53 и т. д. Важно сохранять временные смещения, поскольку, как видно из исходных данных, температура не всегда измерялась через равные промежутки времени.

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

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

Нам на помощь приходят оконные функции.

-- (скопировано)


MIN(Date_time) OVER (
 ORDER BY Date_time ASC ROWS BETWEEN ${window_size} PRECEDING AND CURRENT ROW
) AS min_time,

Это позволит получить дату записи для window_size записей перед ней. Затем для каждой даты и ее min_time будут добавлены все даты и их температуры, которые попадают в диапазон (используя оператор WHERE). Хотя в результате получается несколько копий каждой даты, каждая из них принадлежит отдельному окну, обозначенному min_bound.

#!/user/bin/bash


window_size=30 #НЕ ИЗМЕНЯТЬ без обновления комманды модели
table="temperatures"


min_bound_table="CREATE TABLE IF NOT EXISTS helper
ENGINE = Memory
AS SELECT
   B.min_time AS min_bound,
   A.Date_time,
   A.Air_temp
FROM
(
   SELECT
       Date_time,
       Air_temp
   FROM ${table}
) AS A,
(
   SELECT
       min_time,
       Date_time,
       Air_temp,
       delta,
       a.Date_time
   FROM
   (
       SELECT
           Date_time,
           Air_temp,
           MIN(Date_time) OVER (ORDER BY Date_time ASC ROWS BETWEEN ${window_size} PRECEDING AND CURRENT ROW) AS min_time,
           Date_time - min_time AS delta
       FROM ${table}
   ) AS a
) AS B
WHERE (A.Date_time >= B.min_time) AND (A.Date_time <= B.Date_time)
ORDER BY Date_time ASC"


echo ${min_bound_table} | docker exec -i clickhouse clickhouse-client
#cont ...

После создания вспомогательной таблицы мы можем использовать GROUP BY min_bound, чтобы разделить записи на соответствующие окна, из которых мы затем можем транспонировать строки в один столбец с помощью groupArray (). В результате мы получим массив температур и временных смещений для каждой даты.

CREATE TABLE IF NOT EXISTS model_data
ENGINE = Memory
AS SELECT
   MAX(Date_time) AS time,
   groupArray(Air_temp) AS temp,
   groupArray(Date_time - min_bound) AS delta_t
FROM helper
GROUP BY min_bound
ORDER BY time ASC

Однако мы должны быть осторожны с самыми ранними 30 записями в нашей итоговой таблице: для заполнения их окон не хватило записей, поэтому вместо них была просто продублирована самая ранняя запись. Эти записи можно идентифицировать по нулевым временным смещениям для записей, выходящих за минимальную границу окна.

ALTER TABLE model_data (DELETE WHERE (delta_t[2]) = 0)

Теперь наши данные представлены в подходящем формате для подачи в нашу модель.

Функция стохастической линейной регрессии Clickhouse принимает размер шага, коэффициент регуляризации, размер батча и метод стохастического градиента, затем столбцы целевого значения, а затем каждый параметр. Любой метод, кроме «Adam», дает NaN, а регуляризация, похоже, не оказывает особого влияния. Я выбрал шаг 0.00004, коэффициент 0.15 и размер батча 3.

step=0.00004 # 0.00050
norm=0.15
batch=3 # 20
method="Adam"
train=75000 # количество записей для обучения
make_model="CREATE TABLE IF NOT EXISTS temp_model
ENGINE = Memory
AS SELECT stochasticLinearRegressionState(${step}, ${norm}, ${batch}, '${method}')
(temp[31], temp[1], temp[2], temp[3], temp[4], temp[5], temp[6], temp[7], temp[8], temp[9], temp[10],
temp[11], temp[12], temp[13], temp[14], temp[15], temp[16], temp[17], temp[18], temp[19], temp[20],
temp[21], temp[22], temp[23], temp[24], temp[25], temp[26], temp[27], temp[28], temp[29], temp[30],
delta_t[1], delta_t[2], delta_t[3], delta_t[4], delta_t[5], delta_t[6], delta_t[7], delta_t[8], delta_t[9], delta_t[10],
delta_t[11], delta_t[12], delta_t[13], delta_t[14], delta_t[15], delta_t[16], delta_t[17], delta_t[18], delta_t[19], delta_t[20],
delta_t[21], delta_t[22], delta_t[23], delta_t[24], delta_t[25], delta_t[26], delta_t[27], delta_t[28], delta_t[29], delta_t[30],
delta_t[31], toMonth(time)) AS state
FROM (SELECT temp, delta_t, time FROM model_data ORDER BY time ASC LIMIT ${train})"


echo ${make_model} | docker exec -i clickhouse clickhouse-client

Я также решил включить месяц в качестве параметра из-за вероятной корреляции между месяцем и температурой. В коде примера я обучаю модель только на первых 75 000 из 100 000 записей.

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

run_model="WITH (
       SELECT state
       FROM temp_model
   ) AS model
SELECT
time,
evalMLMethod(model, temp[1], temp[2], temp[3], temp[4], temp[5], temp[6], temp[7], temp[8], temp[9], temp[10],
temp[11], temp[12], temp[13], temp[14], temp[15], temp[16], temp[17], temp[18], temp[19], temp[20],
temp[21], temp[22], temp[23], temp[24], temp[25], temp[26], temp[27], temp[28], temp[29], temp[30],
delta_t[1], delta_t[2], delta_t[3], delta_t[4], delta_t[5], delta_t[6], delta_t[7], delta_t[8], delta_t[9], delta_t[10],
delta_t[11], delta_t[12], delta_t[13], delta_t[14], delta_t[15], delta_t[16], delta_t[17], delta_t[18], delta_t[19], delta_t[20],
delta_t[21], delta_t[22], delta_t[23], delta_t[24], delta_t[25], delta_t[26], delta_t[27], delta_t[28], delta_t[29], delta_t[30],
delta_t[31], toMonth(time)) AS predicted,
   temp[31] AS actual
FROM model_data ORDER BY time DESC LIMIT 200"

echo ${run_model} | docker exec -i clickhouse clickhouse-client

Прогнозируемые значения находятся в середине, а фактические — справа.

Довольно неплохо для этого интервала, хоть и не обошлось без 10-градусного отклонения для 8/7 18:53. Я сохранил все результаты в файл .csv, чтобы позже построить график с помощью следующей команды:

echo "sql command..." | docker exec -i clickhouse clickhouse-client --format CSV> result.csv

Результаты:

Выше приведены прогнозы моей лучшей модели. Ее максимальные и минимальные предсказания были очень точными, а средние отклонения между каждой предсказанной температурой и фактической были почти нулевыми.

Прежде чем показать его график в Excel, вот некоторые из моих предыдущих результатов для контекста:

step=0.00055, norm=0.1, batch=10

Первая попытка

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

step=0.00010, norm=0.15, batch=5, method=”Adam”

Вторая попытка

Этот вариант выглядит лучше. Отклонения менее экстремальны, а прогнозируемые значения следуют общей тенденции.

А теперь моя финальная модель:

step=0.000040, norm=0.15, batch=3, method=”Adam”

Последняя попытка

Отклонения уменьшились, а прогнозы более точно следуют синусоидальному тренду. Если разделить в Excel графики данных из последней попытки, то можно заметить, что пики действительно присутствуют в исходных данных, но модель их просто преувеличивает, как, например, при внезапном провале в июне 2016 года. Не обращайте внимания на значения оси y в этом случае:

На этих трех графиках я уменьшал как размер шага, так и размер батча. Большие размеры шага приводили к диким колебаниям результатов между последовательными прогонами и экстремальным выбросам. Увеличение размера батча, как правило, уменьшало величину этих выбросов, но при этом усредняло прогнозы, в результате чего получалась ровная линия, которая минимально менялась в течение года. (Интересно, что сглаживалась только верхняя граница; минимальные значения продолжали следовать правильной тенденции в более холодные месяцы года). Уменьшение степени «сплющивания» за счет уменьшения размера батча, а также уменьшение величины выбросов за счет уменьшения размера шага позволило получить наиболее точные результаты.

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

Спасибо за ваше время! Надеюсь, вы узнали что-то полезное.

Всё о работе с ClickHouse: от установки и настройки, до продовых решений, можно узнать на онлайн-курсе «ClickHouse для инженеров и архитекторов БД» под руководством экспертов.

© Habrahabr.ru