Baggrundsopgaver på Faust, Del I: Introduktion

Hvordan kom jeg til dette liv?

For nylig arbejdede jeg på backend’en af et højt belastet projekt, hvor jeg skulle organisere regelmæssig udførelse af et stort antal baggrundsopgaver med komplekse beregninger og forespørgsler til tredjepartsservices. Projektet er asynkront, og inden jeg kom ind i det, var der en simpel mekanisme til planlægning af opgaver: en løkke der kontrollerede den nuværende tid og startede grupper af korutiner gennem gather — denne tilgang viste sig at være acceptabel, indtil antallet af korutiner blev titusinder, men da antallet oversteg to tusinde, var vi nødt til at overveje at organisere en normal opgavekø med en broker, flere workere og lignende.

Først besluttede jeg at prøve Celery, som jeg tidligere havde brugt. På grund af projektets asynkronitet dykkede jeg ind i spørgsmålet og så artiklen, samt projekt, skabt af forfatteren af artiklen.

Jeg vil sige, at projektet er meget interessant og fungerer ganske godt i andre applikationer i vores team, og forfatteren selv nævner, at han har været i stand til at implementere det i produktion ved at bruge en asynkron pulje. Men desværre var det ikke helt egnet til mig, da der opstod problemet vedrørende gruppering af opgaveudførelser (se gruppe). På tidspunktet for skrivningen af artiklen issue var allerede afsluttet, men arbejdet blev udført i løbet af en måned. Uanset hvad, ønsker jeg forfatteren held og lykke, da der allerede er fungerende løsninger i biblioteket... Kort sagt, det er mit problem, og for mig var værktøjet lidt for ufuldstændigt. Desuden var der i nogle opgaver 2-3 http-forespørgsler til forskellige tjenester, således at selv ved optimering af opgaverne skaber vi 4.000 tcp-forbindelser cirka hver 2. time — ikke ideelt... Jeg ville gerne have en session pr. opgavetype ved start af workere. Mere om det store antal forespørgsler gennem aiohttp her.

Fordi dette begyndte jeg at søge alternativer og fandt! Skaberne af Celery, mere præcist, som jeg forstod Ask Solem, havde skabt Faust, oprindeligt til projektet robinhood. Faust er inspireret af Kafka Streams og arbejder med Kafka som broker, ligeledes anvendes RocksDB til lagring af resultaterne fra arbejdet med agenter, og det vigtigste er, at biblioteket er asynkront.

Du kan også se en kort sammenligning celery og faust fra skaberne af den seneste: deres forskelle, forskelle mellem mæglere, implementeringen af en simpel opgave. Det er alt sammen ret enkelt, men faust tiltrækker opmærksomhed med en behagelig egenskab — typesikrede data til overførsel til emner.

Hvad skal vi gøre?

Så i en lille serie artikler vil jeg vise, hvordan man indsamler data i baggrundsopgaver med hjælp fra Faust. Kilden til vores eksempelprojekt vil, som titlen antyder, være alphavantage.co. Jeg vil demonstrere, hvordan man skriver agenter (sinks, emner, partitioner), hvordan man udfører regelmæssige (cron) køringer, de mest praktiske cli-kommandoer i faust (en wrapper over click), simpel clustering, og til sidst vil vi integrere datadog (som fungerer ud af boksen) og forsøge at se noget. Til opbevaring af de indsamlede data vil vi bruge mongodb og motor til at forbinde.

P.S. I betragtning af den selvsikkerhed, der er skrevet i afsnittet om overvågning, tror jeg, at læseren i slutningen af den sidste artikel vil se sådan ud:

Krav til projektet

Da jeg allerede har lovet en del, lad os lave en lille liste over, hvad tjenesten skal kunne:

1. Hente værdipapirer og en oversigt over dem (herunder indtægter og udgifter, balance, cash flow — for det seneste år) — regelmæssigt
2. Hente historiske data (for hvert handelsår finde ekstrempriserne for afslutningskurserne) — regelmæssigt
3. Hente de seneste handelsdata — regelmæssigt
4. Hente en konfigureret liste over indikatorer for hvert værdipapir — regelmæssigt

Som sædvanligt vælger vi et navn til projektet fra det blå: horton

Forberede infrastrukturen

Titlen er bestemt stærk, men alt hvad der skal gøres er at skrive en lille konfiguration for docker-compose med kafka (og zookeeper — i en enkelt container), kafdrop (hvis vi vil se beskeder i emnerne), mongodb. Vi får [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) af følgende slags:

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

Der er faktisk ikke noget kompliceret her. For Kafka har vi erklæret to lyttere: en (intern) til brug inden for det samlede netværk og den anden (extern) til forespørgsler udefra, derfor har vi åbnet den udadtil. 2181 er zookeeper-porten. Resten, tænker jeg, er klart.

Vi forbereder projektets skelet

I den grundlæggende version skal strukturen af vores projekt se således ud:

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 *

*Alt det, jeg har markeret rører vi ikke ved endnu, men laver blot tomme filer.

Vi har oprettet strukturen. Nu tilføjer vi de nødvendige afhængigheder, skriver konfigurationen og forbindelsen til MongoDB. Jeg vil ikke medtage hele teksten af filerne i artiklen for ikke at gøre den for lang, så jeg vil lave links til de nødvendige versioner.

Lad os begynde med afhængighederne og metaoplysningerne om projektet — pyproject.toml

Derefter starter vi installationen af afhængighederne og oprettelsen af et virtualenv (eller, du kan selv oprette en mappe venv og aktivere miljøet):

pip3 install poetry (hvis ikke allerede installeret)
poetry install

Lad os nu oprette config.yml — credentialer og hvor vi skal kommunikere. Her kan vi også placere oplysninger til alphavantage. Og så går vi videre til config.py — vi henter data til applikationen fra vores konfiguration. Ja, jeg beklager, jeg har brugt mit eget bibliotek — sitri.

Forbindelsen med MongoDB er heller ikke besværlig. Vi har erklæret en klientklasse til forbindelsen og en grundlæggende klasse til CRUD-operationer, for at gøre det lettere at udføre forespørgsler på samlingerne.

Hvad vil der ske videre?

Artiklen er ikke særlig lang, da jeg kun taler om motivation og forberedelse her, så bær over med mig - jeg lover, at der i næste del vil være action og grafik.

Så i denne næste del skal vi:

1. Skrive en lille klient til alphavantage på aiohttp med forespørgsler til de endepunkter, vi har brug for.
2. Lave en agent, der indsamler data om værdipapirer og historiske priser på dem.

Projektkode

Koden til denne del

Kilde: habr.com