Как упаковать бэкенд-код на Go для аналитики на базе Spark

Всем привет! Меня зовут Ваня Ахлестин, я занимаюсь поддержкой и развитием аналитической платформы кластера Search&Recommendations на базе Spark и Hadoop. 

Большинство сервисов в хайлоаде, работу которых мы логируем и исследуем, давно переписаны на Go. Из-за этого часто необходимо переиспользовать логику сервиса внутри аналитического или ML-приложения на Spark. Как примеры такого кода можно взять расчёт скоров по сложному запросу или ранжирование айтемов для выдачи. 

Реимплементировать и поддерживать несколько вариантов кода дорого, заниматься обстрелом сервисов долго и больно. Поэтому сегодня я расскажу, как начать использовать ваш код из Python или PySpark и не тратить много времени дорогих разработчиков.

Немного о нашей инфраструктуре

Наша платформа построена на классической связке Hadoop/YARN/Spark/PySpark. Основной язык разработки — Python 3.11. Мы стараемся придерживаться баланса между зрелостью и актуальностью софта, чтобы не погрязнуть в поддержке легаси решений. Для этого мы долго обучали разработчиков и аналитиков писать тесты для их задач.

Изначально Python был выбран вместо Scala, чтобы снизить порог входа и дать возможность каждому работать с продуктовым кодом вместо выделенной команды бигдата-инженеров, реимплементирующих код от аналитиков и ML.

В типичных цикл разработки входит:

• Проектирование — обсуждаем и подготавливаем данные для задачи, эксперимента, проговариваем примерную реализацию.

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

• Упаковываем задачу в формат Airflow, настраиваем, покрываем тестами и тестовыми данными, настраиваем метрики.

Для аналитики работы сервисов мы обычно используем пайплайн с логированием JSON структур в Kafka, данные из которой в HDFS поставляет задача на Flink. Реплей таких логов на кластере через PySpark UDF — об этом и написана статья.

CGO, или собираем биндинг дешево

Первое, о чём нужно подумать при работе с кодом сервиса — как организовать биндинг со сложными вложенными структурами вашего сервиса.

Тут есть несколько подходов:

• Довериться генератору биндингов, например GoPy. Самое очевидное решение, которое мы и попробовали в первую очередь. К сожалению, результат был негативным:

  • биндинг содержит много мусора и ненужных артефактов;

  • требуется дополнительно его патчить, не всё работает «из коробки»;

  • структуры биндинга уродливы, над ними нужна ещё одна обертка;

  • код неэффективен;

  • собранный пакет жёстко привязывается к определённой версии интерпретатора — у нас есть несколько версий и дистрибутивов Python.

Поэтому такие биндинги не очень долго прожили — требовали постоянного внимания.

• Написать биндинг вручную. Существует много туториалов, как это сделать, но есть сложности:

  • поддержка однообразных структур и кода на С;

  • есть привязка к ABI Python, нужна сборка под конкретный Python;

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

• Использовать сериализацию в промежуточный формат. Плюсы:

  • если это JSON, то сервис его уже использует, он нативен для кода;

  • Spark и Python уже умеют его использовать «из коробки»;

  • можем передать в код клиента схему с описанием структур;

  • не нужно поддерживать код самого биндинга.

Минус — есть дополнительные расходы на упаковку-распаковку, но и быстрые парсеры тоже есть. На этом варианте мы и остановились.

Минимальный биндинг изнутри

Чтобы было проще разобраться, я подготовил демо-проект со следующей структурой:

Основная идея проекта:

• Собрать shared библиотеку с С-интерфейсом.

• Добавить к ней Python-биндинг с использованием ctypes.

• Упаковать в переносимый пакет (Wheel), готовый для установки в любое окружение для платформы.

Бекэнд-код на Go

Демо-код бэкенд-логики содержит вложенные структуры и код, который их принимает и возвращает — это максимально типичная ситуация в реальной жизни. Структуры уже размечены для JSO- генератора — это необязательно.

package go_backend_lib  type RawDataItem srtuct {    ItemId      int64   'json:"item_id,omitempry"'    Pos         uint8   'json:"pos,omitempty"'    PlatformId  int64   'json:"platform_id,omitempty"'    RegionId    int64   'json:"region_id,omitemrty"'    CTR         float64 'json:"ctr,omitempty"' }  type RawataRecord struct {     Timestamp       int64                  'json:"tm,omitempty"'     X               string                 'json:"x,omitempty"'     Items           []RawDataItem          'json:"items,omitempty"'     UserHistoryItem map[int64]*RawDataIten 'json:"user_history_items,omitempty"'     Tag             string                 'json:"tag,omitempty"' }  type ScoresRecord struct {     Timestamp int64             'json:"tm,omitempty"'     X         string            'json:"x,omitempty"'     Scores    map[int64]float64 'json:"scores,omitempty"' }  func CalcScores(data RawDataRecord) ScorexRecord {     // ... }

Экспортируем функции и структуры через C Call интерфейс

package main  /*  typedef struct Result {   char* value;   char* err; } Result;  */ import "C"  import (     gbl "example.com/go_backend_lib"     "fmt"      jsonschema "github.com/invopop/jsonschema"     jsoniter "github.com/json-iterator/go" )  func ErrStruct(err error) C.Result {     return C.Result{         value: nil,         err:   C.CString(err.Error())     } }  func ResStruct(value string) C.Result {     return C.Result{         value: CCString(value),         err:   nil,     } }

//expert CalcScores func CalcScores(rawDataRecordJson *C.char) C.Result {     jsonBytes := []byte(C.GoString(rawDataRecordJson))     jsonConfig := jsoniter.ConfigCompatibleWithStandardLibrary      record := gbl.RawDataRecord{}     err := jsonConfig.Unmarchal(jsonBytes, &record)     if err != nil {         return ErrStruct(err)     }      scores := gbl.CalcScores(record)     resBytes, err := jsonConfig.Marshal(scores)     if err != nil {         return ErrStruct(err)     }      return ResStruct(string(resBytes[:]))    }  //export GetScoresRecordSchema func GetScoresRecordSchema() C.Result {     schema := jsonschema.Reflect(&gbl.ScoresRecord{})     jsonConfig := jsoniter.ConfigCompatibleWithStandardLibrary     data, err := jsonConfig.Marshal(schema)     if err != nil {         return ErrStruct(err)     }     return ResStruct(string(data)) }  func main () { }

Как видно, всё, что делает код — это принимает и возвращает строки с JSON. Для описания структур используется JSON Schema. Это не идеальный вариант — из-за ограниченности типов JSON больше демонстрация возможности использовать рефлекшены для создания схемы.

Собираем всё в разделяемую библиотеку

'''shell cd ./py_backend_bind/ go build -buildmode=c-shared -o backend_bind/library.so main.go

Обёртка на Python

Используем ctypes с динамической загрузкой собранной библиотеки.

import ctypes import os from sys import platform  __all__ = [     'calc_scores',     'get_raw_data_record_schema',     'get_scores_record_schema', ]  this_dir = os.path.abspath(os.path.dirname(__file__))  if platform == "linux" or platform == "linux2":     suffix = ".so" elif platform == "darwin" :     suffix = ".dylib" else:     raise NotImplementedError()  class Result(ctypes.Structure):     _field_  = [         ('value', ctypes.c_char_p),         ('err', ctypes.c_char_p),     ]  library = ctypes.cdll.LoadLibrary(os.path.join(this_dir, 'library' + suffix))  cs_func = library.CalcScores cs_func.argtypes = [ctypes.c_char_p] cs_func.restype = Result  gsrs_func = library.GetScoresRecordSchema gsrs_func.argtypes = [] gsrs_func.retype = Result

def check_result(res: Result) -> str:     if res.err is not None:         raise RuntimeError(res.err.decode())     return res.value.decode()   def calc_scores(raw_data: str) -> str:     return check_result(cs_func(raw_data.encode()))   def get_scores_record_schema() -> str:     return check_result(gsrs_func())

И не забываем про сборку проекта в пакет:

'''toml [tool.poetry] name = "backend_bind" version = "0.1" description = "Demo for binding GoLang backend lib to python for usage in PySpark Cluster applications" authors = ["Ivan Akhlestin <j.h@g.c>"] packages = [{ include = "backend_bind" }] include = [     { path = "*.so", format = [         "wheel",     ] },     { path = "*.h ", format = [         "wheel",     ] }, ]  [tool.poetry.dependencies] python = ">3.9"  [build-system] requires = ["poetry-core"] build-backend = "poetry.core.masonry.api" 

Всё, что нужно — Python и установленный poetry. При запуске сборки poetry build собирает пакеты source и wheel. Пакет wheel как раз и предназначен для дистрибьюции бинарных зависимостей для Python.

PEX как контейнер, и деплой кода на кластер

Любой PySpark скрипт с python-udf перед тем, как начать выполнять план запроса на кластере, производит сериализацию функции и всех обьектов из её контекста через библиотеку cloudpickle. То есть, для того, чтобы восстановить функцию на кластерных пайтон-воркерах нужно обеспечить действительность всех ссылок из pickle.

Существует несколько способов деплоя Python окружения на кластер:

1. Устанавливать библиотеки в окружение системного Python.

• Плюс: просто сделать.

• Минусы:

  • сложно синхронизировать и следить за всеми артефактами.

  • невозможно работать и отлаживать несколько разных окружений, а использование venv только всё усложняет;

  • сложно отследить, какие версии использовались в определенный момент.

2. Использовать docker/kubernetes.

• Плюс: можно сделать любое окружение.

• Минус: не подходит для bare-metal инсталляций с Hadoop

3. Деплой через EGG.

• Минус: работает только с нативным Python-кодом — не получится использовать биндитнги из-за ограничений zipimport.

• Плюс: работает из коробки, не нужно паковать всё окружение, если библиотека самодостаточна

4. Использование законсервированных окружений в виде conda-pack, venv-pack, pex.

• Плюсы:

  • разово собирается и пакуется окружение, не получится испортить.

  • прозрачно деплоится на кластер, но нужно оркестрировать установку на все тачки.

  • возно версионировать синхронно с основным кодом.

• Минус: PEX-пакет весит как полноценный virtualenv.

Мы используем динамическую версию из Git в виде комбинации тега, майлстоуна, хеша коммита и признака dev-ветки.

Для себя мы выбрали PEX как наиболее зрелый вариант пакета. Если коротко, то это self-extacted virtualenv, упакованный в обычный zip и снабжённый бустстап-секцией для поиска хост-интерпретатора. 

Для использования PEX как интерпретатора Python на кластере нужно модифицировать настройки Spark. У нас конфигурация контекста выглядит примерно так:

''' "spark.yarn.dist.files": "hdfs://apps/cluster_code/cluster_code-24.2.dev91+adab58fe.pex", "spark.executorEnv.PYSPARK_Python": f"./cluster_code-24.2.dev91+addab58fe.pex", "spark.executorEnv.PEX_ROOT": "./.pex" "spark.executorEnv.PYSPARK_DRIVER_PYTHON": "/opt/python3.11/bin/python3", "spark.pyspark.python": f"./cluster_code-24.2.dev91+adab58fe.pex", '''

Для сборки PEX-пакета достаточно самой утилиты pex и списка всех пакетов, которые должны быть в него установлены:

'''shell # в папочку dist складываем все внутренние пакеты, в том числе и биндинг backend_bind pex -repo./dust cluster_code==24.2.dev91+adab58fe pyspark==3.5.1 backend_bind==0.01 -f dist/ -o ./dist/cluster_code-24.2.dev91+adab58fe.pex '''

Естественно, что в PEX-пакете должен находится pyspark, иначе Spark не сможет запустить Python-воркеры на кластерных машинах.

Включаем нашу UDF в работу

Подразумевается, что Spark-сессия запущена, настроена и используется PEX.

''' import pyspark.sql.function as F import backend_bind  logs_df = spark.read.text('/logs/backend_service/2024-05-01') calc_scores_udf = F.udf(backend_bind.calc_scores, 'string') scores_df = logs_df.select(calc_scores_sdf(F.col('value'))) scores_df.take(10) '''

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

Что почитать по теме

• Туториал по CGo.

• Документация к ctypes.

• PySpark Package Management.