🥇Taustatööd Fausti abil, Osa I: Sissejuhatus

Kuidas ma sellisele elule jõudsin?

Hiljaaegu pidin töötama väga koormatud projekti tagaplaanil, kus tuli korraldada regulaarne suurte arvu taustatööde täitmine keeruliste arvutuste ja kolmandate osapoolte teenuste päringutega. Projekt on asünkroonne ja enne minu tulekut oli seal lihtne krondimehhanism ülesannete käivitamiseks: tsükkel, mis kontrollis praegust aega ja käivitas rühmades korutiine gather kaudu — see lähenemine osutus vastuvõetavaks seni, kuni selliseid korutiine oli kümneid ja sadu, kuid kui nende kogus ületas kaks tuhat, tuli hakata mõtlema korraliku ülesannete järjekorra, maakleri, mitme töötaja ja muude asjade korraldamise peale.

Alguses otsustasin proovida Celeryt, mida olin varem kasutanud. Seoses projekti asünkroonsusega uputasin end küsimusse ja nägin artiklile, samuti projekt, autori loodud.

Ütleksin nii, et projekt on väga huvitav ja töötab üsna edukalt ka meie meeskonna teistes rakendustes. Isegi autor räägib, et on suutnud selle tootmisse viia, kasutades asünkroonset pulti. Kuid kahjuks ei sobinud see mulle hästi, kuna selgus, et probleem gruppeeritud ülesannete käivitamine (vt grupp). Artikli kirjutamise hetkel probleem oli juba lõpetatud, kuid töö käidi kuu aega. Igatahes, autorile edu ja kõike head, kuna tegu on juba töötavate funktsioonidega… kokkuvõttes, asi on minusse ja see tööriist tundus mulle veel toores. Lisaks oli mõnedes ülesannetes 2-3 http-pära erinevates teenustes, seega isegi ülesannete optimeerimise korral loome 4 tuhat tcp ühendust umbes iga 2 tunni tagant — mitte just ideaalne… Tahaksin seadistada ühe seansi ühe ülesande tüübi jaoks tööde käivitamisel. Natuke rohkem informatsiooni aiohttp kaudu tehtud suurte päringute arvust. siin.

Sellega seoses hakkasin otsima alternatiive ja leidsin! Celery looja, nagu ma aru sain, oli Ask Solem, kes lõi Faust, algselt projekti jaoks robinhood. Faust on kirjutatud Kafka Streams'i inspiratsioonist ja töötab Kafka'ga kui vahendajaga, samuti kasutatakse tulemuste salvestamiseks rocksdb-d. Peamine eelis on see, et raamatukogu on asünkroonne.

Samuti võite vaadata lühikest võrdlust celery ja faust'i vahel, mille on loonud viimase autorid: nende erinevused, vahendajate erinevused, lihtsa ülesande teostamine. Kõik on üsna lihtne, kuid fausti juures tõmbab tähelepanu meeldiv omadus — tüübistatud andmed teema edastamiseks.

Mida me teeme?

Nii et väikeses artiklite seerias näitan, kuidas koguda andmeid taustategevustes Faust'i abil. Meie näidisprojekti allikaks on, nagu nimigi ütleb, alphavantage.co. Demonstreerin, kuidas kirjutada agente (sink, teemad, partitsioonid), kuidas teha regulaarset (cron) täitmist, mugavaid faust'i cli-käske (wrapper klikiga) ja lihtsat klasterdamist. Lõpuks ühendame datadogi (mis töötab kohe) ja proovime midagi näha. Kogutud andmete salvestamiseks kasutame mongodb-d ja motorit ühendamiseks.

P.S. Nähtavasti on jällegi kirjutatud usaldusväärselt jälgimise punkt, arvan, et lugedes viimase artikli lõppu, näeb lugeja tõenäoliselt välja umbes nii:

Projekti nõuded

Seoses sellega, et olen juba lubanud, koostame väikese nimekirja sellest, mida teenus peaks suutma:

Kandma välja väärtpaberid ja ülevaade nende kohta (sealhulgas kasumid ja kahjumid, bilanss, rahavoog — viimase aasta jooksul) — regulaarselt
Kandma välja ajaloolised andmed (iga kauplemisaasta kohta leidma sulgemishinna ekstreemumid) — regulaarselt
Kandma välja viimased kauplemisandmed — regulaarselt
Kandma välja konfigureeritud indikaatorite nimekiri iga väärtpaberi jaoks — regulaarselt

Nagu ikka, valime projekti nime täiesti suvaliselt: horton

Valmistame infrastruktuuri ette

Pealdis on muidugi tugev, kuid kõik, mida tuleb teha, on kirjutada väike konfiguraator docker-compose jaoks koos kafka (ja zookeeper — ühes konteineris), kafdrop (kui soovime vaadata sõnumeid teemas), mongodb. Saame [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) järgmise vormi:

versioon: '3'

teenused:
  db:
    konteineri_nimi: horton-mongodb-local
    pilt: mongo:4.2-bionic
    käsk: mongod --port 20017
    taaskäivitamine: alati
    pordid:
      - 20017:20017
    keskkond:
      - MONGO_INITDB_DATABASE=horton
      - MONGO_INITDB_ROOT_USERNAME=admin
      - MONGO_INITDB_ROOT_PASSWORD=admin_password

  kafka-service:
    konteineri_nimi: horton-kafka-local
    pilt: obsidiandynamics/kafka
    taaskäivitamine: alati
    pordid:
      - "2181:2181"
      - "9092:9092"
    keskkond:
      KAFKA_LISTENERS: "INTERNEER //:29092, VÄLJASPOOL //:9092"
      KAFKA_ADVERTISED_LISTENERS: "INTERNEER //kafka-service:29092, VÄLJASPOOL //localhost:9092"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNEER:PLAINTEXT, VÄLJASPOOL:PLAINTEXT"
      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNEER"
      KAFKA_ZOOKEEPER_SESSION_TIMEOUT: "6000"
      KAFKA_RESTART_ATTEMPTS: "10"
      KAFKA_RESTART_DELAY: "5"
      ZOOKEEPER_AUTOPURGE_PURGE_INTERVAL: "0"

  kafdrop:
    konteineri_nimi: horton-kafdrop-local
    pilt: 'obsidiandynamics/kafdrop:latest'
    taaskäivitamine: alati
    pordid:
      - '9000:9000'
    keskkond:
      KAFKA_BROKERCONNECT: kafka-service:29092
    sõltub:
      - kafka-service

Siin pole midagi keerulist. Kafka jaoks on määratud kaks kuulajat: üks (sisemine) kasutamiseks koosoleku võrgu sees, ja teine (väline) väljastpoolt päringute jaoks, seega suunati see välja. 2181 on zookeeper'i port. Ülejäänud kohta arvan, et on selge.

Valmistame projekti karkassi

Baasmudelis peaks meie projekti struktuur välja nägema järgmiselt:

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 *

*Kõik, mida ma märkisin hetkel ei puuduta, vaid loome lihtsalt tühjad failid.**

Loome struktuuri. Nüüd lisame vajalikud sõltuvused, kirjutame konfiguratsiooni ja ühenduse mongodb-ga. Täielikke failitekste ma artiklis ei too, et mitte venitada, vaid teen lingid vajalikele versioonidele.

Alustame sõltuvustest ja projekti metaandmetest — pyproject.toml

Edasi, käivitame sõltuvuste installimise ja virtualenv loomise (või võite ise luua kausta venv ja aktiveerida keskkonna):

pip3 install poetry (kui pole veel installitud)
poetry install

Nüüd loome config.yml — mandaadid ja kuhu minna. Seal võib samuti paigutada andmed alphavantage'ile. Nüüd liigume edasi config.py — eraldame rakenduse andmed meie konfiguratsioonist. Jah, tunnistan, et kasutasin oma teeki — sitri.

MongoDB-ga ühendamiseks on kõik väga lihtne. Meil on klassi klient ühendamiseks ja põhiklasse krudode jaoks, et oleks lihtsam teha päringuid kogumitele.

Mis juhtub edasi?

Artikkel tuli mitte väga suur, kuna räägin siin ainult motivatsioonist ja ettevalmistusest, seega palun vabandust – luban, et järgmises osas tuleb tegevust ja graafikat.

Nii et järgmises osas teeme järgmist:

Kirjutame väikese kliendi alphavantage'i jaoks aiohttp'ga, tehes päringuid vajalikele lõpp-punktidele.
Loome agendi, mis kogub andmeid väärtpaberite kohta ja ajaloolisi hindu nende kohta.

Projekti kood

Selle osa kood

Allikas: habr.com