Pozadinské úlohy na Fauste, Časť I: Úvod

Ako som sa dostal k tomuto životu?

Nedávno som musel pracovať na backend-e vysoko zaťaženého projektu, v ktorom bolo potrebné organizovať pravidelné vykonávanie veľkého množstva pozadových úloh so zložitými výpočtami a požiadavkami na externé služby. Projekt je asynchrónny a predtým, než som prišiel, mal jednoduchý mechanizmus cron spúšťania úloh: cyklus s kontrolou aktuálneho času a spúšťanie skupín korutín cez gather — tento prístup sa ukázal ako akceptovateľný, pokiaľ takých korutín bolo desiatky a stovky, avšak keď ich počet prekročil dve tisíce, musel som premýšľať o organizácii normálnej fronty úloh s brokerom, niekoľkými pracovníkmi a ďalšími.

Najprv som sa rozhodol vyskúšať Celery, ktorým som predtým používal. V súvislosti s asynchrónnosťou projektu som sa ponoril do otázky a videl článok, a taktiež projekt, vytvorený autorom článku.

Poviem tak, projekt je veľmi zaujímavý a celkom úspešne funguje v iných aplikáciách nášho tímu, a aj sám autor hovorí o tom, že sa mu podarilo nasadiť do produkcie, použijúc asynchrónny pool. Ale, bohužiaľ, to sa mi veľmi nehodilo, pretože sa objavila probléma s skupinovým spúšťaním úloh (pozri. skupina). V čase písania článku issue už bola uzavretá, avšak práca sa vykonávala počas mesiaca. V každom prípade prajem autorovi veľa šťastia a všetko dobré, pretože funkčné veci na knižnici už sú… celkovo, problém bol vo mne a pre mňa sa ukázal nástroj príliš surový. Navyše, v niektorých úlohách bolo po 2-3 http požiadavky na rôzne služby, takže aj pri optimalizácii úloh vytvárame 4 tisíc tcp spojení, približne každé 2 hodiny — nie veľmi… Rád by som vytváral reláciu na jeden typ úloh pri spúšťaní pracovníkov. Trochu podrobnejšie o veľkom počte požiadaviek cez aiohttp tu.

V súvislosti s tým som začal hľadať alternatívy a našiel! Tvorcovia Celery, a konkrétne, ako som pochopil Ask Solem, vytvorili Faust, pôvodne pre projekt robinhood. Faust je napísaný pod vplyvom Kafka Streams a pracuje s Kafka ako brokerom, tiež pre ukladanie výsledkov z práce agentov sa používa rocksdb, a najdôležitejšie — táto knižnica je asynchrónna.

Tiež si môžete pozrieť stručné porovnanie celery a faust od tvorcov posledného: ich rozdiely, rozdiely brokerov, realizáciu elementárnej úlohy. Všetko je celkom jednoduché, avšak v faust zaujme príjemná vlastnosť — typované dáta na prenos do topiku.

Čo budeme robiť?

Takže v malej sérii článkov ukážem, ako zbierať dáta v pozadí pomocou Faust. Zdrojem pre náš príkladový projekt bude, ako naznačuje názov, alphavantage.co. Predvediem, ako písať agentov (sink, topiky, partitiony), ako vykonávať pravidelné (cron) spúšťanie, najpohodlnejšie cli príkazy faust (obalka okolo click), jednoduché klastrovanie, a na konci pripojíme datadog (fungujúci out-of-the-box) a pokúsime sa niečo vidieť. Na uchovávanie zozbieraných dát budeme používať mongodb a motor na pripojenie.

P.S. Na základe istej istoty, s ktorou je napísaný bod o monitorovaní, si myslím, že čitateľ na konci posledného článku bude vyzerať asi takto:

Požiadavky na projekt

Vzhľadom na to, čo som už stihol sľúbiť, zostavíme malý zoznam toho, čo by mal servis vedieť:

1. Pravidelne vyvážať cenné papiere a prehľad o nich (vrátane ziskov a strát, bilancie, cash flow — za posledný rok)
2. Pravidelne vyvážať historické dáta (pre každý obchodný rok zistiť extrémy ceny zatvorenia obchodov)
3. Pravidelne vyvážať najnovšie obchodné dáta
4. Pravidelne vyvážať nastavený zoznam indikátorov pre každý cenný papier

Ako sa patrí, vyberáme názov projektu zo vzduchu: horton

Pripravujeme infraštruktúru

Názov je síce silný, avšak všetko, čo treba urobiť — je napísať malú konfiguráciu pre docker-compose s kafka (a zookeeper — v jednom kontejnere), kafdrop (ak sa nám zažiada pozrieť správy v topikoch), mongodb. Získavame [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) nasledujúceho typu:

verzia: '3'

služby:
  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 je nič zložité. Pre Kafka boli definované dva listeneri: jeden (interný) na používanie vo vnútri kompozitnej siete a druhý (externý) na požiadavky zvonka, preto sme ho posunuli von. 2181 – port Zookeepera. O ostatnom si myslím, že je jasné.

Pripravujeme kostru projektu

V základnej variante by štruktúra nášho projektu mala vyzerať takto:

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 *

*Všetko, čo som označil teraz zatiaľ nerušíme, len vytvárame prázdne súbory.**

Vytvorili sme štruktúru. Teraz pridáme potrebné závislosti, napíšeme konfiguráciu a pripojenie k MongoDB. Kompletný text súborov nebudem uvádzať v článku, aby som to nepredlžoval, a poskytnem odkazy na potrebné verzie.

Začneme so závislosťami a metadátami projektu — pyproject.toml

Potom spustíme inštaláciu závislostí a vytvorenie virtualenv (alebo, môžete si sami vytvoriť priečinok venv a aktivovať prostredie):

pip3 install poetry (ak ešte nie je nainštalované)
poetry install

Teraz vytvoríme config.yml — kredity a kam sa pripojiť. Hneď tam môžeme umiestniť aj dáta pre Alphavantage. A prechádzame k config.py — extrahujeme dáta pre aplikáciu z našej konfigurácie. Áno, priznávam, použil som svoju knižnicu — sitri.

Čo sa týka pripojenia k MongoDB — je to úplne jednoduché. Definovali sme triedu klienta pre pripojenie a základnú triedu pre CRUD operácie, aby bolo jednoduchšie robiť požiadavky na kolekcie.

Čo sa bude diať ďalej?

Článok nie je veľmi dlhý, pretože tu hovorím iba o motivácii a príprave, takže sa nehnevajte — sľubujem, že v ďalšej časti bude akcia a grafika.

Takže v tejto najbližšej časti urobíme:

1. Napíšeme jednoduchého klienta pre alphavantage na aiohttp s požiadavkami na potrebné koncové body.
2. Urobíme agenta, ktorý bude zbierať údaje o cenných papieroch a historické ceny na nich.

Kód projektu

Kód tejto časti

Zdroj: habr.com