Hintergrundaufgaben auf Faust, Teil I: Einführung

Wie bin ich zu diesem Leben gekommen?

Kürzlich musste ich an einem stark ausgelasteten Backend-Projekt arbeiten, bei dem es notwendig war, regelmäßig eine große Anzahl von Hintergrundaufgaben mit komplexen Berechnungen und Anfragen an externe Dienste zu organisieren. Das Projekt ist asynchron, und bevor ich kam, gab es einen einfachen Mechanismus zum Kicken von Aufgaben: eine Schleife mit der Überprüfung der aktuellen Zeit und dem Start von Gruppen von Koroutinen über gather — dieser Ansatz war akzeptabel, solange die Anzahl der Koroutinen in den Zehner- und Hundertbereich fiel, jedoch als ihre Zahl über zweitausend stieg, musste ich über die Organisation einer normalen Aufgabenwarteschlange mit einem Broker, mehreren Workern und anderem nachdenken.

Zunächst habe ich beschlossen, Celery auszuprobieren, das ich zuvor verwendet hatte. Angesichts der Asynchronität des Projekts tauchte ich in das Thema ein und sah den Artikel, sowie Projekt, erstellt vom Autor des Artikels.

Ich sage mal so, das Projekt ist sehr interessant und funktioniert ziemlich gut in anderen Anwendungen unseres Teams. Der Autor selbst erwähnt, dass er es erfolgreich in Produktion gebracht hat, indem er einen asynchronen Pool genutzt hat. Leider war das jedoch nicht ganz passend für mich, da sich herausstellte, ein Problem mit dem gruppierten Start von Aufgaben (siehe group). Zum Zeitpunkt der Artikelverfassung war das Problem bereits behoben, jedoch dauerte die Arbeit einen Monat. Auf jeden Fall wünsche ich dem Autor viel Erfolg und alles Gute, da es bereits funktionierende Lösungen in der Bibliothek gibt… im Grunde liegt es an mir, und das Tool erschien mir noch nicht ausgereift. Außerdem gab es in einigen Aufgaben 2-3 HTTP-Anfragen an verschiedene Dienste, wodurch wir selbst bei der Optimierung der Aufgaben 4000 TCP-Verbindungen alle zwei Stunden erzeugen – nicht ideal… Ich würde gerne eine Sitzung für einen Typ von Aufgaben beim Starten der Worker erzeugen. Ein wenig mehr über die große Anzahl von Anfragen über aiohttp. hier.

In diesem Zusammenhang begann ich, nach Alternativen zu suchen und fand! Von den Erstellern von Celery, speziell, wie ich verstanden habe, Ask Solem, wurde Faustgeschaffen, ursprünglich für das Projekt robinhood. Faust wurde von Kafka Streams inspiriert und verwendet Kafka als Broker. Für die Speicherung der Ergebnisse der Agenten setzen wir rocksdb ein. Das Wichtigste ist jedoch, dass die Bibliothek asynchron ist.

Außerdem können Sie einen Blick auf eine kurze Vergleichsanalyse von Celery und Faust, erstellt von den Entwicklern von Faust, werfen: ihre Unterschiede, die Broker-Differenzen und die Implementierung einer grundlegenden Aufgabe. Alles ist ziemlich einfach, aber besonders interessant an Faust ist die Funktion der typisierten Daten, die an das Topic übergeben werden.

Was werden wir tun?

In einer kleinen Artikelsammlung zeige ich, wie man mit Faust Daten in Hintergrundaufgaben sammelt. Die Quelle für unser Beispielprojekt wird, wie der Titel schon sagt, alphavantage.co. Ich werde demonstrieren, wie man Agenten (Sink, Topics, Partitionen) erstellt, regelmäßige (Cron) Ausführungen einrichtet, die nützlichen CLI-Befehle von Faust (eine Wrapper über Click) verwendet, einfaches Clustering durchführt und am Ende Datadog (aus der Box funktionierend) anbindet, um zu sehen, was wir erreichen können. Für die Speicherung der gesammelten Daten verwenden wir MongoDB und Motor für die Verbindung.

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

Projektanforderungen

Da ich bereits einige Versprechungen gemacht habe, erstellen wir eine kleine Liste dessen, was der Service können sollte:

1. Regelmäßig wertpapierbezogene Daten und einen Überblick darüber extrahieren (einschließlich Gewinne und Verluste, Bilanz, Cashflow – für das letzte Jahr)
2. Regelmäßig historische Daten extrahieren (für jedes Handelsjahr die Extrempreise der Schlusskurse finden)
3. Regelmäßig die letzten Handelsdaten extrahieren
4. Regelmäßig eine konfigurierte Liste von Indikatoren für jedes Wertpapier extrahieren

Wie es sich gehört, wählen wir einen Projektnamen willkürlich aus: horton

Infrastruktur vorbereiten

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

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

Es ist wirklich nichts kompliziert. Für Kafka wurden zwei Listener definiert: einer (intern) für die Nutzung innerhalb des Composite-Netzwerks und der andere (extern) für Anfragen von außen, weshalb er nach außen geleitet wurde. 2181 ist der Port des Zookeepers. Darüber hinaus ist alles klar.

Wir bereiten das Grundgerüst des Projekts vor.

Im Basisversion 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 stellen die Verbindung zu MongoDB her. Die vollständigen Inhalte der Dateien werde ich in diesem Artikel nicht auflisten, um es nicht unnötig in die Länge zu ziehen, sondern ich werde Links zu den benötigten Versionen einfügen.

Beginnen wir mit den Abhängigkeiten und den Metadaten zum Projekt — pyproject.toml

Danach starten wir die Installation der Abhängigkeiten und die Erstellung eines virtualenv (alternativ können Sie auch selbst einen Ordner venv erstellen und die Umgebung aktivieren):

pip3 install poetry (falls noch nicht installiert)
poetry install

Jetzt erstellen wir config.yml — die Credentials und wo wir uns verbinden. Dort können auch die Daten für Alphavantage hinterlegt werden. Nun 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 zu MongoDB ist ganz einfach. Wir haben eine Client-Klasse für die Verbindung und eine Basisklasse für Skripte, um Abfragen zu Sammlungen zu erleichtern.

Wie geht es weiter?

Der Artikel ist nicht besonders lang, da ich hier nur über Motivation und Vorbereitung spreche. Bitte entschuldigt, ich verspreche, dass im nächsten Teil Action und Grafiken kommen werden.

Also, im nächsten Teil werden wir:

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

Projektcode

Code dieses Teils

Quelle: habr.com