ფონური ამოცანები ფაუსტზე, ნაწილი I: შესავალი

როგორ მომივიდა ასე ცხოვრება?

არც ისე დიდი ხნის წინ მომიწია მუშაობა უაღრესად დატვირთული პროექტის ფონზე, რომელშიც საჭირო იყო დიდი რაოდენობით ფონური დავალების რეგულარული შესრულების ორგანიზება რთული გამოთვლებით და მესამე მხარის სერვისების მოთხოვნით. პროექტი ასინქრონულია და სანამ მე მოვიდოდი, მას ქონდა ამოცანების გაშვების მარტივი მექანიზმი: მარყუჟი, რომელიც ამოწმებს მიმდინარე დროს და აწარმოებს კორუტინის ჯგუფებს შეკრების საშუალებით - ეს მიდგომა მისაღები აღმოჩნდა მანამ, სანამ არ იყო ათობით და ასობით ასეთი კორუტინი. თუმცა, როცა მათმა რაოდენობამ ორ ათასს გადააჭარბა, მომიწია ფიქრი ბროკერთან, რამდენიმე მუშაკთან და ა.შ. ჩვეულებრივი დავალების რიგის მოწყობაზე.

ჯერ გადავწყვიტე გამომეცადა ნიახური, რომელსაც ადრე ვიყენებდი. პროექტის ასინქრონული ხასიათის გამო ჩავუღრმავდი კითხვას და ვნახე მუხლი, და პროექტი, შექმნილი სტატიის ავტორის მიერ.

ამას ვიტყვი, პროექტი ძალიან საინტერესოა და საკმაოდ წარმატებით მუშაობს ჩვენი გუნდის სხვა აპლიკაციებში და თავად ავტორი ამბობს, რომ ასინქრონული აუზის გამოყენებით შეძლო მისი წარმოებაში გატანა. მაგრამ, სამწუხაროდ, ეს ნამდვილად არ მომეწონა, როგორც აღმოჩნდა პრობლემა დავალებების ჯგუფური გაშვებით (იხ. ჯგუფი). წერის დროს საკითხი უკვე დაკეტილია, თუმცა მუშაობა ერთი თვეა მიმდინარეობს. ყოველ შემთხვევაში წარმატებები ავტორს და ყოველივე კარგს, რადგან ლიბზე უკვე მუშაობს საქმეები... ზოგადად, საქმე ჩემშია და ხელსაწყო ნესტიანი აღმოჩნდა ჩემთვის. გარდა ამისა, ზოგიერთ ამოცანას ჰქონდა 2-3 http მოთხოვნა სხვადასხვა სერვისებზე, ასე რომ, დავალების ოპტიმიზაციის დროსაც ვქმნით 4 ათას TCP კავშირს, დაახლოებით ყოველ 2 საათში - არც თუ ისე კარგი... მე მინდა შევქმნა სესია ერთი ტიპისთვის. დავალება მუშების დაწყებისას. ცოტა მეტი მოთხოვნის დიდი რაოდენობის შესახებ aiohttp-ის საშუალებით აქ.

ამ მხრივ დავიწყე ძებნა ალტერნატივები და იპოვა! ნიახურის შემქმნელები, კონკრეტულად, როგორც მე მესმის ჰკითხე სოლემს, იყო შექმნილი ფაუსტი, თავდაპირველად პროექტისთვის robinhood. ფაუსტი შთაგონებულია Kafka Streams-ით და მუშაობს კაფკასთან, როგორც ბროკერი, rocksdb ასევე გამოიყენება აგენტების მუშაობის შედეგების შესანახად და რაც მთავარია ბიბლიოთეკა ასინქრონულია.

ასევე, შეგიძლიათ ნახოთ სწრაფი შედარება ნიახური და ფაუსტი ამ უკანასკნელის შემქმნელებისგან: მათი განსხვავებები, განსხვავება ბროკერებს შორის, ელემენტარული დავალების შესრულება. ყველაფერი საკმაოდ მარტივია, თუმცა ყურადღებას იპყრობს ფაუსტის სასიამოვნო ფუნქცია - აკრეფილი მონაცემები თემაზე გადასაცემად.

Რას ვაკეთებთ?

ასე რომ, სტატიების მოკლე სერიაში მე გაჩვენებთ, თუ როგორ უნდა შეაგროვოთ მონაცემები ფონური ამოცანებიდან ფაუსტის გამოყენებით. ჩვენი მაგალითის პროექტის წყარო იქნება, როგორც სახელიდან ჩანს, alphavantage.co. მე ვაჩვენებ როგორ დავწეროთ აგენტები (ნიჟარები, თემები, ტიხრები), როგორ გავაკეთოთ რეგულარული (cron) შესრულება, ყველაზე მოსახერხებელი faust cli ბრძანებები (შეფუთვა დაწკაპუნებაზე), მარტივი კლასტერირება და ბოლოს დავამაგრებთ მონაცემთა ძაღლს ( ყუთში მუშაობა) და სცადეთ რაიმეს დანახვა. შეგროვებული მონაცემების შესანახად ჩვენ გამოვიყენებთ mongodb-ს და ძრავას დასაკავშირებლად.

P.S. თუ ვიმსჯელებთ იმ ნდობით, რომლითაც დაიწერა აბზაცი მონიტორინგის შესახებ, ვფიქრობ, რომ ბოლო სტატიის ბოლოს მკითხველი მაინც ასე გამოიყურება:

პროექტის მოთხოვნები

იმის გამო, რომ მე უკვე დავპირდი, მოდით შევადგინოთ მცირე ჩამონათვალი იმისა, თუ რა უნდა გააკეთოს სერვისმა:

1. ატვირთეთ ფასიანი ქაღალდები და მათი მიმოხილვა (მათ შორის მოგება და ზარალი, ბალანსი, ფულადი სახსრები - გასული წლის განმავლობაში) - რეგულარულად
2. ატვირთეთ ისტორიული მონაცემები (თითოეული სავაჭრო წლისთვის, იპოვეთ ვაჭრობის დახურვის ფასის უკიდურესი მნიშვნელობები) - რეგულარულად
3. ატვირთეთ უახლესი სავაჭრო მონაცემები - რეგულარულად
4. ატვირთეთ ინდიკატორების მორგებული სია თითოეული უსაფრთხოებისთვის - რეგულარულად

როგორც მოსალოდნელი იყო, ჩვენ ვირჩევთ პროექტის სახელს ნულიდან: ჰორტონი

ვამზადებთ ინფრასტრუქტურას

სათაური, რა თქმა უნდა, ძლიერია, თუმცა, ყველაფერი რაც თქვენ უნდა გააკეთოთ, არის დაწეროთ პატარა კონფიგურაცია docker-compose-ისთვის კაფკათი (და ზოოპარკი - ერთ კონტეინერში), კაფდროპი (თუ გვინდა შევხედოთ შეტყობინებებს თემებში), mongodb. ვიღებთ [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) შემდეგი ფორმის:

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

აქ საერთოდ არაფერია რთული. კაფკაზე გამოცხადდა ორი მსმენელი: ერთი (შიდა) კომპოზიციური ქსელის შიგნით გამოსაყენებლად და მეორე (გარე) გარედან მოთხოვნისთვის, ასე რომ, მათ გადაუგზავნეს იგი გარეთ. 2181 - ზოოპარკის პორტი. დანარჩენი, ვფიქრობ, გასაგებია.

პროექტის ჩონჩხის მომზადება

ძირითად ვერსიაში ჩვენი პროექტის სტრუქტურა ასე უნდა გამოიყურებოდეს:

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 *

*ყველაფერი რაც აღვნიშნე ჩვენ ჯერ არ ვეხებით, უბრალოდ ვქმნით ცარიელ ფაილებს.**

ჩვენ შევქმენით სტრუქტურა. ახლა დავამატოთ საჭირო დამოკიდებულებები, ჩავწეროთ კონფიგურაცია და დავუკავშირდეთ mongodb-ს. სტატიაში არ მოგაწოდებთ ფაილების სრულ ტექსტს, რომ არ დავაგვიანო, მაგრამ მივცემ ბმულებს საჭირო ვერსიებზე.

დავიწყოთ დამოკიდებულებებითა და მეტა პროექტის შესახებ - pyproject.toml

შემდეგი, ჩვენ ვიწყებთ დამოკიდებულებების ინსტალაციას და ვირტუალურის შექმნას (ან შეგიძლიათ თავად შექმნათ venv საქაღალდე და გაააქტიუროთ გარემო):

pip3 install poetry (если ещё не установлено)
poetry install

ახლა შევქმნათ config.yml - რწმუნებათა სიგელები და სად დავაკაკუნო. თქვენ შეგიძლიათ დაუყოვნებლივ განათავსოთ მონაცემები alphavantage-ისთვის. კარგი, მოდით გადავიდეთ config.py - ამოიღეთ მონაცემები აპლიკაციისთვის ჩვენი კონფიგურაციიდან. დიახ, ვაღიარებ, გამოვიყენე ჩემი lib - სატირი.

მონგოსთან დაკავშირებისას ყველაფერი საკმაოდ მარტივია. გამოაცხადა კლიენტის კლასი დასაკავშირებლად და საბაზო კლასი cruds-ისთვის, რათა გაადვილდეს კითხვების დასმა კოლექციებზე.

რა მოხდება შემდეგ?

სტატია არც თუ ისე გრძელია, რადგან აქ მხოლოდ მოტივაციაზე და მომზადებაზე ვსაუბრობ, ამიტომ ნუ დამაბრალებთ - გპირდებით, რომ შემდეგ ნაწილს ექნება მოქმედება და გრაფიკა.

ასე რომ, ამ მომდევნო ნაწილში ჩვენ:

1. მოდით დავწეროთ პატარა კლიენტი alphavantage-ისთვის aiohttp-ზე ჩვენთვის საჭირო ბოლო წერტილების მოთხოვნით.
2. მოდით შევქმნათ აგენტი, რომელიც შეაგროვებს მონაცემებს ფასიანი ქაღალდების და მათთვის ისტორიული ფასების შესახებ.

პროექტის კოდი

კოდი ამ ნაწილისთვის

წყარო: www.habr.com