Hintergrundaufgaben in Faust, Teil I: Einleitung

Wie bin ich zu diesem Leben gekommen?

Vor nicht allzu langer Zeit musste ich an dem Backend eines stark ausgelasteten Projekts arbeiten, in dem viele Hintergrundaufgaben mit komplexen Berechnungen und Anfragen an externe Dienste regelmäßig ausgeführt werden sollten. Das Projekt ist asynchron und bevor ich kam, gab es einen einfachen Mechanismus zum Auslösen von Aufgaben über Cron: einen Zyklus, der die aktuelle Zeit überprüft und Gruppen von Koroutinen über gather startet - dieses Vorgehen war akzeptabel, solange es nur Dutzende oder Hunderte von Koroutinen gab, aber als die Anzahl über zweitausend stieg, musste ich darüber nachdenken, wie man eine ordentliche Aufgabenwarteschlange mit einem Broker, mehreren Workern und mehr organisiert.

Zunächst beschloss ich, Celery auszuprobieren, das ich früher verwendet hatte. Aufgrund der Asynchronität des Projekts tauchte ich in das Thema ein und sah einen Artikel, sowie Projekt, das vom Autor des Artikels erstellt wurde.

Ich kann sagen, das Projekt ist sehr interessant und funktioniert ziemlich gut in anderen Anwendungen unseres Teams, und auch der Autor selbst sagt, dass er es in Produktion bringen konnte, indem er einen asynchronen Pool verwendete. Leider war dies jedoch nicht ganz passend für mich, da ich entdeckte Problem bei der gruppenmäßigen Ausführung von Aufgaben (siehe Gruppe). Zum Zeitpunkt des Schreibens des Artikels issue bereits geschlossen, aber die Arbeit wurde einen Monat lang fortgesetzt. Auf jeden Fall wünsche ich dem Autor viel Glück und alles Gute, denn die funktionierenden Dinge in der Bibliothek sind bereits vorhanden… im Grunde genommen liegt das an mir und für mich war das Werkzeug noch zu unreif. Zudem gab es in einigen Aufgaben 2-3 HTTP-Anfragen an verschiedene Dienste, so dass wir selbst bei Optimierung der Aufgaben etwa 4.000 TCP-Verbindungen alle 2 Stunden erstellen - nicht ideal... Ich hätte gerne bei der Ausführung der Worker eine Sitzung für einen Aufgabentyp erstellt. Ein wenig mehr über die große Anzahl von Anfragen über aiohttp hier.

In diesem Zusammenhang begann ich nach Alternativen zu suchen und fand sie! Die Schöpfer von Celery, und zwar, wie ich verstand Ask Solem, schufen Faust, ursprünglich für das Projekt robinhood. Faust wurde inspiriert von Kafka Streams und arbeitet mit Kafka als Broker, außerdem wird für die Speicherung der Ergebnisse der Agentenarbeit rocksdb verwendet, und das Wichtigste ist, dass die Bibliothek asynchron ist.

Zusätzlich können Sie sich das ansehen kurzes Vergleich Celery und Faust von den Machern des Letzten: ihre Unterschiede, Unterschiede bei Brokern, die Implementierung einfacher Aufgaben. Alles ist ziemlich einfach, jedoch zieht bei Faust eine angenehme Besonderheit die Aufmerksamkeit auf sich – typisierte Daten zur Übertragung in ein Topic.

Was sollen wir tun?

In einer kleinen Artikelreihe werde ich zeigen, wie man Daten in Hintergrundaufgaben mit Faust sammelt. Die Quelle für unser Beispielprojekt wird, wie aus dem Titel ersichtlich, alphavantage.co. Ich werde demonstrieren, wie man Agenten (Sink, Topics, Partitionen) schreibt, regelmäßige (Cron) Ausführungen macht, die praktischsten CLI-Befehle von Faust (eine Wrapper über Click), einfaches Clustering, und am Ende werden wir Datadog (der out-of-the-box funktioniert) einbinden und versuchen, etwas zu sehen. Zum Speichern der gesammelten Daten verwenden wir MongoDB und Motor zur Verbindung.

P.S. Angesichts der Zuversicht, mit der der Punkt über Monitoring geschrieben ist, denke ich, dass der Leser am Ende des letzten Artikels so aussehen wird:

Anforderungen an das Projekt

Im Hinblick auf das, was ich bereits versprochen habe, lassen Sie uns eine kleine Liste dessen erstellen, was der Service können sollte:

1. Wertpapiere und einen Überblick über sie (einschließlich Gewinne und Verluste, Bilanz, Cashflow – im letzten Jahr) regelmäßig exportieren
2. Historische Daten (für jedes Handelsjahr Extremwerte der Schlusskurse finden) – regelmäßig exportieren
3. Aktuelle Handelsdaten exportieren – regelmäßig
4. Eine konfigurierbare Liste von Indikatoren für jedes Wertpapier exportieren – regelmäßig

Wie es sich gehört, wählen wir einen Namen für das Projekt willkürlich aus: Horton

Die Infrastruktur vorbereiten

Die Überschrift ist natürlich stark, aber alles, was zu tun ist – ist eine kleine Konfiguration für docker-compose mit Kafka (und Zookeeper – in einem Container), Kafdrop (falls wir Nachrichten in den Topics sehen möchten), MongoDB schreiben. Wir erhalten [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) in folgender Art:

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

Hier ist überhaupt nichts kompliziert. Für Kafka wurden zwei Listener deklariert: einer (intern), um innerhalb des zusammengesetzten Netzwerks zu verwenden, und der zweite (extern), um Anfragen von außen zu bearbeiten, weshalb er nach außen weitergeleitet wurde. 2181 ist der Port des Zookeepers. Für den Rest denke ich, ist alles klar.

Wir bereiten das Skelett des Projekts vor

Im Grundsatz sollte die Struktur unseres Projekts wie folgt aussehen:

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 *

*Alles, was ich markiert habe berühren wir vorerst nicht, sondern erstellen einfach leere Dateien.**

Wir haben die Struktur erstellt. Jetzt fügen wir die notwendigen Abhängigkeiten hinzu, schreiben die Konfiguration und die Verbindung zu MongoDB. Den vollständigen Text der Dateien werde ich im Artikel nicht wiedergeben, um es nicht zu lang zu machen, sondern mache stattdessen Links zu den benötigten Versionen.

Lass uns mit den Abhängigkeiten und den Metadaten zum Projekt beginnen - pyproject.toml

Dann starten wir die Installation der Abhängigkeiten und die Erstellung des Virtualenv (oder, alternativ könnt ihr den Ordner venv selbst erstellen und die Umgebung aktivieren):

pip3 install poetry (wenn noch nicht installiert)
poetry install

Jetzt erstellen wir config.yml — die Anmeldedaten und den Verbindungsort. Dort können wir auch die Daten für Alphavantage hinterlegen. Und jetzt gehen wir zu config.py — wir extrahieren die Daten für die Anwendung aus unserer Konfiguration. Ja, ich gebe es zu, ich habe meine eigene Bibliothek verwendet — sitri.

Die Verbindung mit MongoDB ist ganz einfach. Wir haben erklärt die Client-Klasse zur Verbindung und die Basisklasse für die CRUD-Operationen, um die Anfragen an die Sammlungen zu erleichtern.

Was wird als Nächstes passieren?

Der Artikel ist nicht sehr lang geworden, da ich hier nur über Motivation und Vorbereitung spreche. Also bitte verzeihe mir – ich verspreche, dass es im nächsten Teil Action und Grafiken geben wird.

Also, in diesem nächsten Teil werden wir:

1. Einen kleinen Client für alphavantage mit aiohttp schreiben, der Anfragen an die benötigten Endpunkte sendet.
2. Einen Agenten erstellen, der Daten über Wertpapiere und deren historische Preise sammelt.

Der Projektcode

Der Code dieses Teils

Quelle: habr.com