Bakgrundsuppgifter om Faust, del I: Introduktion

Hur kom jag hit?

För inte så länge sedan fick jag arbeta med backend på ett högt belastat projekt där det var nödvändigt att organisera regelbundet utförande av ett stort antal bakgrundsuppgifter med komplexa beräkningar och anrop till externa tjänster. Projektet är asynkront och innan jag kom dit användes en enkel cron-mekanism för att starta uppgifter: en loop som kontrollerade aktuell tid och startade grupper av korutiner med gather — detta tillvägagångssätt visade sig vara acceptabelt så länge det fanns tiotals eller hundratals korutiner, men när deras antal översteg två tusen, var jag tvungen att tänka på hur man skulle organisera en normal uppgiftskö med en broker, flera arbetare och så vidare.

Först beslutade jag mig för att prova Celery, som jag tidigare hade använt. Med tanke på projektets asynkronitet, fördjupade jag mig i frågan och såg artikeln, och också projektet, skapad av artikelns författare.

Låt mig säga så här, projektet är mycket intressant och fungerar ganska bra i andra applikationer av vårt team, och författaren själv säger att han lyckades få det i produktion, genom att använda en asynkron pool. Men tyvärr, det passade inte mig så bra, eftersom det upptäcktes problem med gruppräkningsuppgifter (se. grupp). Vid tidpunkten för artikelns skrivande problem var den redan stängd, men arbetet pågick under en månad. Hur som helst, jag önskar författaren lycka till och allt gott, eftersom det redan finns fungerande verktyg i biblioteket... i allmänhet, det handlar om mig och för mig var verktyget lite för grönt. Dessutom, i vissa uppgifter gjorde jag 2-3 http-anrop till olika tjänster, så även vid optimering av uppgifterna skapar vi 4 000 tcp-anslutningar varje 2 timmar — inte så bra... Jag skulle vilja skapa en session för en typ av uppgifter när jag startar arbetarna. Lite mer om det stora antalet förfrågningar genom aiohttp. ). Ett typiskt exempel på användning:.

I samband med detta började jag leta efter alternativ och jag hittade! Skaparna av Celery, och specifikt, som jag förstod Ask Solem, skapade Faust, ursprungligen för projektet robinhood.Faust är skriven inspirerad av Kafka Streams och fungerar med Kafka som broker, dessutom används rocksdb för att lagra resultat från agenternas arbete, och viktigast av allt — biblioteket är asynkront.

Dessutom kan du ta en titt på en kort jämförelse celery och faust från skaparna av den senaste: deras skillnader, skillnader i mäklare, implementeringen av en grundläggande uppgift. Allt är ganska enkelt, men faust har en trevlig egenskap - typade data för överföring till en topic.

Vad ska vi göra?

Så i en liten artikelserie kommer jag att visa hur man samlar in data i bakgrundsuppgifter med hjälp av Faust. Källan för vårt exempelprojekt kommer, som namnet antyder, alphavantage.co. Jag kommer att demonstrera hur man skriver agenter (sink, topics, partitioner), hur man schemalägger (cron) körningar, de mest användarvänliga cli-kommandona i faust (ett wrapper över click), enkel klustring, och i slutet ansluter vi datadog (som fungerar direkt ur lådan) och försöker se något. För att lagra de insamlade uppgifterna kommer vi att använda mongodb och motor för anslutning.

P.S. Baserat på den säkerhet med vilken punkten om övervakning skrivits, tror jag att läsaren i slutet av den sista artikeln kommer att se ut ungefär så här:

Krav på projektet

Med tanke på vad jag redan har lovat, låt oss göra en liten lista över vad tjänsten ska kunna:

1. Exportera värdepapper och översikt över dem (inklusive vinster och förluster, balans, kassaflöde - för det senaste året) - regelbundet
2. Exportera historiska data (för varje handelsår hitta extremvärden för stängningspriser) - regelbundet
3. Exportera de senaste handelsdata - regelbundet
4. Exportera en anpassad lista med indikatorer för varje värdepapper - regelbundet

Som det bör vara väljer vi ett namn för projektet från ingenstans: horton

Förbereda infrastrukturen

Rubriken är förstås stark, men allt vi behöver göra är att skriva en liten konfiguration för docker-compose med kafka (och zookeeper - i en behållare), kafdrop (om vi vill se meddelanden i topicarna), mongodb. Vi får [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) följande typ:

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

Det är helt enkelt. För kafka har vi deklarerat två lyssnare: en (intern) för användning inom det sammansatta nätverket och den andra (extern) för förfrågningar utifrån, därför har vi exponerat den. 2181 är zookeepers port. När det kommer till resten, tror jag att det är klart.

Vi förbereder projektets struktur

I grundversionen ska strukturen av vårt projekt se ut så här:

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 *

*Allt jag markerat rör vi oss inte vid ännu, utan vi skapar bara tomma filer.

Vi har skapat strukturen. Nu lägger vi till de nödvändiga beroendena, skriver konfigurationen och kopplingen till mongodb. Jag kommer inte att inkludera hela texten för filerna i artikeln för att inte dra ut på tiden, utan jag kommer att länka till de nödvändiga versionerna.

Låt oss börja med beroendena och metainformationen om projektet— pyproject.toml

Därefter utför vi installationen av beroendena och skapar virtualenv (eller så kan ni själva skapa en mapp för venv och aktivera miljön):

pip3 install poetry (om det inte redan är installerat)
poetry install

Nu skapar vi config.yml — kredentialer och vart vi ska ansluta. Dit kan vi också placera data för alphavantage. Och nu går vi vidare till config.py — vi extraherar data för applikationen från vår konfiguration. Ja, jag erkänner, jag använde mitt bibliotek— sitri.

När det gäller anslutningen till mongo är det väldigt enkelt. Vi har deklarerat en klientklass för anslutning och en basklass för CRUD-operationer, så att det blir enklare att göra förfrågningar på kollektionerna.

Vad händer härnäst?

Artikeln är inte så lång, eftersom jag här bara pratar om motivation och förberedelse, så snälla ha förståelse — jag lovar att det kommer att bli action och grafik i nästa del.

Så, i denna nästa del kommer vi att:

1. Skriva en liten klient för alphavantage på aiohttp med förfrågningar till de endpoints vi behöver.
2. Skapa en agent som samlar in data om värdepapper och historiska priser för dem.

Projektkod

Koden för denna del

Källa: habr.com