Pozadí úloh na Faust, Část I: Úvod

Jak jsem se dostal k tomu, co dělám?

Nedávno jsem musel pracovat na backendu vysoce zatíženého projektu, kde bylo potřeba zajistit pravidelné vykonávání velkého množství pozadí úloh se složitými výpočty a dotazy na externí služby. Projekt je asynchronní a předtím, než jsem přišel, měl jednoduchý mechanismus pro spouštění úloh pomocí cron: cyklus, který kontroloval aktuální čas a spouštěl skupiny korutin pomocí gather – tento přístup byl přijatelný, dokud bylo takových korutin desítky a stovky, nicméně když jejich počet překročil dvě tisíce, bylo nutné uvažovat o organizaci normální fronty úloh s brokerem, několika pracovníky a dalšími.

Nejprve jsem se rozhodl vyzkoušet Celery, které jsem dříve používal. Vzhledem k asynchronnosti projektu jsem se ponořil do problematiky a viděl článek, a také projekt, vytvořenou autorem článku.

Řeknu to takto, projekt je velmi zajímavý a docela úspěšně funguje v jiných aplikacích našeho týmu, a sám autor říká, že se mu podařilo nasadit do produkce, využívaje asynchronní pool. Ale, bohužel, pro mě to moc nevyhovovalo, protože se objevily problém problémy se skupinovým spouštěním úloh (viz group). V době psaní článku issue byla již uzavřena, avšak práce probíhala po dobu jednoho měsíce. V každém případě, autorovi přeji hodně štěstí a vše nejlepší, protože funkční věci na knihovně už existují… obecně, chyba je ve mně a pro mě byl nástroj příliš nedokončený. Navíc, v některých úlohách bylo 2-3 http dotazy na různé služby, takže i při optimalizaci úloh vytváříme 4 tisíce TCP spojení zhruba každé 2 hodiny — což není ideální… Chtěl bych vytvářet relaci na jeden typ úloh při spuštění pracovníků. Trochu podrobněji o velkém počtu dotazů přes aiohttp tady.

V tomto ohledu jsem začal hledat alternativy a našel! Tvůrci celery, konkrétně, jak jsem pochopil Ask Solem, vytvořili Faust, původně pro projekt robinhood. Faust je napsán pod vlivem Kafka Streams a pracuje s Kafkou jako broker, také pro ukládání výsledků z práce agentů se používá rocksdb, a co je nejdůležitější — je, že knihovna je asynchronní.

Také se můžete podívat na stručné srovnání celery a faust od tvůrců poslední: jejich rozdíly, rozdíly brokerů, realizaci elementární úlohy. Všechno je to docela jednoduché, nicméně, v faust zaujme příjemná vlastnost - typizovaná data pro přenos do topiku.

Co budeme dělat?

A takže, v malé sérii článků ukážu, jak shromažďovat data v pozadí pomocí Faust. Zdroj pro náš ukázkový projekt bude, jak už název napovídá, alphavantage.co. Předvedu, jak psát agenty (sink, topiky, partitiony), jak provádět pravidelnou (cron) exekuci, velmi užitečné cli-commands faust (obálka nad click), jednoduché clustering a nakonec připojíme datadog (funkční out of the box) a pokusíme se něco vidět. Pro ukládání shromážděných dat budeme používat mongodb a motor pro připojení.

P.S. Soudě podle jistoty, s níž byl napsán bod o monitorování, myslím, že čtenář na konci posledního článku bude vypadat, takto:

Požadavky na projekt

Vzhledem k tomu, že jsem už dost sliboval, vytvořme malý seznam toho, co by měl servis umět:

1. Pravidelně stahovat cenné papíry a přehled o nich (včetně zisků a ztrát, bilance, cash flow - za poslední rok)
2. Pravidelně stahovat historická data (pro každý obchodní rok najít extrémy uzavíracích cen obchodů)
3. Pravidelně stahovat poslední obchodní data
4. Pravidelně stahovat nastavený seznam indikátorů pro každý cenný papír

Jak se patří, vybereme jméno projektu od stropu: horton

Připravujeme infrastrukturu

Název je samozřejmě silný, nicméně, vše, co je třeba udělat - je napsat malou konfiguraci pro docker-compose s kafka (a zookeeper - v jednom kontejneru), kafdrop (pokud budeme chtít vidět zprávy v topikech), mongodb. Získáme [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) následujícího typu:

verze: '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

Tady není vůbec nic složitého. Pro kafku byly vyhlášeny dva listenery: jeden (interní) pro použití uvnitř složené sítě a druhý (externí) pro požadavky zvenčí, takže jsme ho přesměrovali ven. 2181 — port zookeepera. Jinak si myslím, že je vše jasné.

Připravujeme strukturu projektu

V základní variantě by struktura našeho projektu měla vypadat 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šechno, co jsem označil prozatím necháváme, a jednoduše vytváříme prázdné soubory.**

Vytvořili jsme strukturu. Teď přidáme potřebné závislosti, napíšeme konfiguraci a připojíme se k mongodb. Kompletní text souborů v článku nebudu uvádět, abych to nezdržoval, a odkazoval na potřebné verze.

Začneme se závislostmi a metadaty projektu — pyproject.toml

Dále spustíme instalaci závislostí a vytvoření virtualenv (nebo si můžete sami vytvořit složku venv a aktivovat prostředí):

pip3 install poetry ( pokud ještě není nainstalováno)
poetry install

Teď vytvoříme config.yml — přihlašovací údaje a kam se připojit. Hned sem můžeme umístit i data pro alphavantage. A teď přejdeme k config.py — extrakci dat pro aplikaci z naší konfigurace. Ano, přiznávám, použil jsem svou knihovnu — sitri.

Pokud jde o připojení k mongu — to je naprosto jednoduché. Vyhlásili jsme třídu klienta pro připojení a základní třídu pro CRUD operace, aby bylo snadnější provádět dotazy na kolekce.

Co bude dál?

Článek není příliš dlouhý, protože zde hovořím pouze o motivaci a přípravě, takže se nezlobte — slibuji, že v další části budou akce a grafika.

Takže, v té nejbližší další části my:

1. Napíšeme malého klienta pro alphavantage na aiohttp s dotazy na potřebné koncové body.
2. Učiníme agenta, který bude sbírat data o cenných papírech a historické ceny za ně.

Kód projektu

Kód této části

Zdroj: habr.com