Zadania w tle na Faust, Część I: Wprowadzenie

Jak doszedłem do takiego życia?

Niedawno przyszło mi pracować nad backendem wysoko obciążonego projektu, w którym trzeba było zorganizować regularne wykonywanie ogromnej liczby zadań w tle złożonymi obliczeniami oraz zapytaniami do zewnętrznych usług. Projekt jest asynchroniczny, a zanim przyszedłem, mieli tam prosty mechanizm cron do uruchamiania zadań: pętla sprawdzająca bieżący czas i uruchamiająca grupy korutyn przez gather — takie podejście okazało się akceptowalne do momentu, gdy liczba tych korutyn osiągnęła kilkadziesiąt lub setki, jednak kiedy ich liczba przekroczyła dwa tysiące, trzeba było pomyśleć o zorganizowaniu normalnej kolejki zadań z brokerem, kilkoma workerami i innymi rzeczami.

Na początku postanowiłem przetestować Celery, z którego korzystałem wcześniej. W związku z asynchronicznością projektu, zagłębiłem się w temat i zobaczyłem artykuł, a także projekt, stworzoną przez autora artykułu.

Powiem tak, projekt jest bardzo ciekawy i całkiem skutecznie działa w innych aplikacjach naszego zespołu, a sam autor mówi, że udało mu się wprowadzić do produkcji, używając asynchronicznego poola. Niestety, mi to nie bardzo pasowało, ponieważ odkryłem problem problem z grupowym uruchamianiem zadań (zob. grupa). W momencie pisania artykułu issue już była zamknięta, jednak prace trwały przez miesiąc. W każdym razie, życzę powodzenia autorowi i wszystkiego dobrego, ponieważ działające rozwiązania w bibliotece już istnieją… ogólnie rzecz biorąc, problem leży po mojej stronie, a dla mnie narzędzie okazało się zbyt surowe. Dodatkowo, w niektórych zadaniach było po 2-3 zapytania http do różnych usług, w ten sposób nawet przy optymalizacji zadań tworzymy 4 tysiące połączeń tcp, mniej więcej co 2 godziny — to nie jest zbyt dobre… Chciałbym tworzyć sesje dla jednego typu zadań podczas uruchamiania workerów. Nieco więcej o dużej liczbie zapytań przez aiohttp tutaj.

W związku z tym zacząłem szukać alternatyw i znalazłem! Twórcami celery, a konkretnie, jak zrozumiałem Ask Solem, stworzono Faust, pierwotnie dla projektu robinhood. Faust jest napisany pod wrażeniem Kafka Streams i działa z Kafka jako broker, a także do przechowywania wyników pracy agentów używa rocksdb, a co najważniejsze — biblioteka jest asynchroniczna.

Możesz również zobaczyć krótkie porównanie Celery i Faust od twórców ostatniej wersji: ich różnice, różnice pomiędzy brokerami, realizacja podstawowego zadania. Wszystko jest dość proste, jednak w Faście przyciąga uwagę miła cecha — typizowane dane do przesyłania do topicu.

Co zamierzamy zrobić?

Otóż w małej serii artykułów pokażę, jak zbierać dane w zadaniach w tle za pomocą Fausta. Źródłem dla naszego przykładowego projektu będzie, jak wskazuje nazwa, alphavantage.co. Zademonstruję, jak pisać agenty (sink, topiki, partycje), jak realizować regularne (cron) zadania, wygodne komendy CLI Fausta (oprócz click), prosty klastering, a na koniec podłączymy Datadog (działający od razu) i spróbujemy coś zobaczyć. Do przechowywania zebranych danych użyjemy MongoDB i Motor do połączenia.

P.S. Sądzę, że z pewnością, z jaką napisano punkt o monitorowaniu, czytelnik na końcu ostatniego artykułu powinien wyglądać właśnie tak:

Wymagania dotyczące projektu

W związku z tym, co już obiecałem, sporządźmy krótki spis rzeczy, które powinien umieć serwis:

1. Regularnie eksportować papiery wartościowe i przegląd na ich temat (w tym zyski i straty, bilans, przepływ gotówki — za ostatni rok)
2. Regularnie eksportować dane historyczne (dla każdego roku handlowego znajdować ekstremalne wartości ceny zamknięcia)
3. Regularnie eksportować najnowsze dane handlowe
4. Regularnie eksportować skonfigurowaną listę wskaźników dla każdego papieru wartościowego

Jak to bywa, wybieramy nazwę projektu na chybił trafił: horton

Przygotowanie infrastruktury

Nagłówek jest z pewnością mocny, jednak wszystko, co trzeba zrobić, to napisać małą konfigurację dla docker-compose z Kafka (i Zookeeper w jednym kontenerze), Kafdrop (jeśli będziemy chcieli zobaczyć wiadomości w topikach), MongoDB. Otrzymujemy [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) w następującej formie:

version: '3'

services:
  db:
    container_name: horton-mongodb-local
    image: mongo:4.2-bionic
    command: mongod --port 20017
    restart: always
    ports:
      - 20017:20017
    environment:
      - MONGO_INITDB_DATABASE=horton
      - MONGO_INITDB_ROOT_USERNAME=admin
      - MONGO_INITDB_ROOT_PASSWORD=admin_password

  kafka-service:
    container_name: horton-kafka-local
    image: obsidiandynamics/kafka
    restart: always
    ports:
      - "2181:2181"
      - "9092:9092"
    environment:
      KAFKA_LISTENERS: "INTERNAL://:29092,EXTERNAL://:9092"
      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka-service:29092,EXTERNAL://localhost:9092"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"
      KAFKA_ZOOKEEPER_SESSION_TIMEOUT: "6000"
      KAFKA_RESTART_ATTEMPTS: "10"
      KAFKA_RESTART_DELAY: "5"
      ZOOKEEPER_AUTOPURGE_PURGE_INTERVAL: "0"

  kafdrop:
    container_name: horton-kafdrop-local
    image: 'obsidiandynamics/kafdrop:latest'
    restart: always
    ports:
      - '9000:9000'
    environment:
      KAFKA_BROKERCONNECT: kafka-service:29092
    depends_on:
      - kafka-service

Tu nie ma nic skomplikowanego. Dla kafka zdefiniowano dwa listener'y: jeden (internal) do użycia wewnątrz sieci kompozytowej, a drugi (external) do zapytań z zewnątrz, dlatego został on udostępniony na zewnątrz. 2181 to port zookeeper'a. W pozostałych sprawach myślę, że jest jasne.

Przygotowujemy szkielet projektu

W podstawowej wersji struktura naszego projektu powinna wyglądać tak:

horton
├── docker-compose.yml
└── horton
    ├── agents.py *
    ├── alphavantage.py *
    ├── app.py *
    ├── config.py
    ├── database
    │   ├── connect.py
    │   ├── cruds
    │   │   ├── base.py
    │   │   ├── __init__.py
    │   │   └── security.py *
    │   └── __init__.py
    ├── __init__.py
    ├── records.py *
    └── tasks.py *

*Wszystko, co oznaczyłem na razie nie ruszamy, a po prostu tworzymy puste pliki.**

Stworzyliśmy strukturę. Teraz dodamy potrzebne zależności, napiszemy konfigurację i połączenie z mongodb. Pełnego tekstu plików nie podam w artykule, aby nie przedłużać, a zamieszczę linki do potrzebnych wersji.

Zacznijmy od zależności i metadanych projektu — pyproject.toml

Następnie uruchamiamy instalację zależności i tworzenie virtualenv (lub możesz samodzielnie stworzyć folder venv i aktywować środowisko):

pip3 install poetry (jeśli jeszcze nie zainstalowane)
poetry install

Teraz stworzymy config.yml — dane i miejsce połączenia. Można od razu umieścić tam również dane dla alphavantage. A teraz przechodzimy do config.py — wyciągamy dane dla aplikacji z naszego pliku konfiguracyjnego. Tak, przyznaję się, użyłem swojej biblioteki — sitri.

W przypadku połączenia z Mongo — to całkiem proste. Zdefiniowano klasę klienta do połączenia oraz podstawową klasę do CRUD-ów, aby łatwiej było robić zapytania do kolekcji.

Co będzie dalej?

Artykuł nie jest zbyt długi, ponieważ mówię tylko o motywacji i przygotowaniu, więc proszę o wybaczenie — obiecuję, że w następnej części będzie akcja i grafika.

A więc, w tej kolejnej części:

1. Napiszemy małą aplikację kliencką dla alphavantage na aiohttp z zapytaniami do potrzebnych nam punktów końcowych.
2. Stworzymy agenta, który będzie zbierał dane o papierach wartościowych i ich historyczne ceny.

Kod projektu

Kod tej części

Źródło: habr.com