Háttérfeladatok a Faustnál, I. rész: Bevezetés

Hogyan jutottam el idáig?

Nemrégiben egy nagy terhelésű projekt háttéralkalmazásán dolgoztam, ahol számos háttérfeladat rendszeres végrehajtása volt szükséges, bonyolult számításokkal és külső szolgáltatások hívásaival. A projekt aszinkron, és mielőtt érkeztem, egy egyszerű cron-alapú feladatindító mechanizmus működött: egy ciklus, amely ellenőrzi a jelenlegi időt, és csoportos coroutine-okat indít a gather segítségével — ez a megközelítés addig volt elfogadható, amíg a coroutine-ok száma néhány tucat és száz között mozgott, de amikor túllépte a kétezer számot, el kellett kezdenem gondolkodni egy normális feladatkezelő rendszer megszervezésén brókerrel, több munkás és egyebekkel.

Először a Celeryt akartam kipróbálni, amit korábban használtam. A projekt aszinkronitása miatt mélyebben belemerültem a kérdésbe, és láttam cikket, valamint projekt, a cikk szerzője által létrehozva.

Azt tudom mondani, hogy a projekt nagyon érdekes, és meglehetősen sikeresen működik csapatunk más alkalmazásaiban, és maga a szerző is azt mondja, hogy sikerült élesben bevezetni egy aszinkron pult használatával. De sajnos, nekem ez nem igazán vált be, mivel felmerült probléma a feladatok csoportindításával kapcsolatban (ld. csoport). A cikk írásának időpontjában issue már lezárult, azonban a munka egy hónapig tartott. Mindenesetre, a szerzőnek sok szerencsét és minden jót kívánok, mivel a könyvtárban már vannak működő dolgok... összességében ez az én hibám, és nekem a nála található eszköz még kicsit nyers. Ráadásul, egyes feladatokban 2-3 http-kérést is végeztem különböző szolgáltatásokhoz, így még a feladatok optimalizálásával is 4000 tcp kapcsolatot hozunk létre körülbelül 2 óránként — nem túl jó... Jó lenne, ha a munkások indításakor egy típusú feladatoknál létrehozhatnánk egy munkamenetet. Kicsit részletesebben a aiohttp-n keresztül megvalósított nagy számú kérdésekről itt.

Ebből kifolyólag elkezdtem keresni alternatívákat és találtam! A Celery készítői, konkrétan, ahogy megértettem Ask Solem, létrehozták Faust, eredetileg a robinhoodprojekt számára. A Faust a Kafka Streams ihletésével készült, és a Kafkát használja brókerként, a feladatügynökök munkájának eredményeinek tárolásához rocksdb-t használnak, és ami a legfontosabb — a könyvtár aszinkron.

Továbbá, nézze meg a rövid összehasonlítást A zeller és a faust az utolsó készítőitől: különbségeik, a brókerek eltérései, alapvető feladatok megvalósítása. Mindez rendkívül egyszerű, de a faust-nál figyelemre méltó egy kellemes jellemző — a típusos adatok átadása a topikba.

Mit fogunk tenni?

Tehát egy kisebb cikk sorozatban megmutatom, hogyan gyűjthetjük az adatokat háttérfeladatokban a Faust segítségével. A példa projektünk forrása, amint a cím is sugallja, alphavantage.co. Megmutatom, hogyan kell ügynököket írni (sink, topikok, partíciók), hogyan végezzük el a rendszeres végrehajtást (cron), a faust legkönnyebb cli parancsait (wrap a click köré), egyszerű klaszterezést, és a végén integrálni fogjuk a datadog-ot (amely dobozból működik) és megpróbálunk valamit látni. Az összegyűjtött adatok tárolásához a mongodb-t és a motort fogjuk használni a csatlakozáshoz.

P.S. A monitorozásról szóló rész írásának magabiztosságát látva, gondolom, hogy az olvasó a legutolsó cikk végén így fog kinézni:

A projekt követelményei

Mivel már megígértem egyet és mást, készítsünk egy rövid listát arról, mit kell tudnia a szolgáltatásnak:

1. Részvények és azok áttekintésének exportálása (beleértve a nyereséget és veszteséget, mérleg, cash flow — az utolsó évre) — rendszeresen
2. Történelmi adatok exportálása (minden kereskedési évre a záró árak szélsőségeit keresni) — rendszeresen
3. Az utolsó kereskedési adatok exportálása — rendszeresen
4. Minden részvényhez beállított indikátorlistának az exportálása — rendszeresen

Ahogy az illik, a projektnek nevet választunk a levegőből: horton

Az infrastruktúra előkészítése

A cím természetesen erős, de mindössze annyit kell tennünk, hogy írunk egy kis konfigurációt a docker-compose-hoz a kafka-hoz (és a zookeeper-hez — egy konténerben), kafdrop-ot (ha szeretnénk megnézni az üzeneteket a topikokban), a mongodb-t. A következő formátumot kapjuk [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) a következő formában:

verzió: '3'

szolgáltatások:
  db:
    konténer_neve: horton-mongodb-local
    kép: mongo:4.2-bionic
    parancs: mongod --port 20017
    újraindítás: mindig
    portok:
      - 20017:20017
    környezet:
      - MONGO_INITDB_DATABASE=horton
      - MONGO_INITDB_ROOT_USERNAME=admin
      - MONGO_INITDB_ROOT_PASSWORD=admin_password

  kafka-szolgáltatás:
    konténer_neve: horton-kafka-local
    kép: obsidiandynamics/kafka
    újraindítás: mindig
    portok:
      - "2181:2181"
      - "9092:9092"
    környezet:
      KAFKA_LISTENERS: "BELSŐ://:29092, KÜLSŐ://:9092"
      KAFKA_ADVERTISED_LISTENERS: "BELSŐ://kafka-szolgáltatás:29092, KÜLSŐ://localhost:9092"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "BELSŐ:PLAINTEXT, KÜLSŐ:PLAINTEXT"
      KAFKA_INTER_BROKER_LISTENER_NAME: "BELSŐ"
      KAFKA_ZOOKEEPER_SESSION_TIMEOUT: "6000"
      KAFKA_RESTART_ATTEMPTS: "10"
      KAFKA_RESTART_DELAY: "5"
      ZOOKEEPER_AUTOPURGE_PURGE_INTERVAL: "0"

  kafdrop:
    konténer_neve: horton-kafdrop-local
    kép: 'obsidiandynamics/kafdrop:latest'
    újraindítás: mindig
    portok:
      - '9000:9000'
    környezet:
      KAFKA_BROKERCONNECT: kafka-szolgáltatás:29092
    függőségek:
      - kafka-szolgáltatás

Itt igazából semmi bonyolult nincs. A kafka számára két listener-t definiáltunk: egyet (belsőt) a kompozit hálózaton belüli használatra, és a másikat (külsőt) a külső kérésekhez, ezért azt ki is toltuk. 2181 a zookeeper portja. A többi szerintem világos.

Készítjük a projekt vázát

Alapértelmezett esetben a projektünk struktúrája így kell, hogy kinézzen:

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

*Mindent, amit megjelöltem most még nem érintünk, csak üres fájlokat hozunk létre.**

Létrehoztuk a struktúrát. Most adjuk hozzá a szükséges függőségeket, írjunk konfigurációt és csatlakozzunk a mongodb-hez. A fájlok teljes szövegét nem fogom megadni a cikkben, hogy ne húzzam el, inkább hivatkozom a szükséges verziókra.

Kezdjük a függőségekkel és a projekt metaadataival — pyproject.toml

Ezután indítsuk el a függőségek telepítését és a virtualenv létrehozását (vagy készíthettek saját venv mappát és aktiválhatjátok a környezetet):

pip3 install poetry (ha még nincs telepítve)
poetry install

Most létrehozunk egy config.yml — krediteket és azt, hová érjük el. Azonnal be lehet helyezni az alphavantage adatait is. Nos, áttérünk a config.py — kinyerjük az adatokat az alkalmazás számára a konfigurációnkból. Igen, beismerem, használtam a saját könyvtáramat — sitri.

A mongodb-hez való csatlakozás nagyon egyszerű. Beállítottunk egy ügyféltípust a csatlakozáshoz és egy alap osztályt a CRUD-hoz, hogy könnyebb legyen lekérdezéseket végezni a gyűjteményekkel.

Mi következik ezután?

A cikk nem túlságosan hosszú, mivel itt csak a motivációról és az előkészületekről beszélek, úgyhogy ne haragudjatok – ígérem, a következő részben már akció és grafika is lesz.

Tehát, a következő részben:

1. Írunk egy kis klienst az alphavantage-hoz aiohttp-val, a szükséges végpontokra küldött lekérdezésekkel.
2. Készítünk egy ügynököt, amely adatokat gyűjt a értékpapírokról és a történelmi áraikról.

A projekt kódja

Ennek a résznek a kódja

Forrás: habr.com