🥇Tugas latar belakang Faust, Bagian I: Pendahuluan

Bagaimana saya bisa hidup seperti ini?

Belum lama ini saya harus mengerjakan bagian belakang proyek yang sarat muatan, di mana perlu mengatur pelaksanaan rutin sejumlah besar tugas latar belakang dengan perhitungan rumit dan permintaan layanan pihak ketiga. Proyek ini tidak sinkron dan sebelum saya datang, ia memiliki mekanisme sederhana untuk tugas peluncuran cron: sebuah loop yang memeriksa waktu saat ini dan meluncurkan grup coroutine melalui pengumpulan - pendekatan ini ternyata dapat diterima sampai ada lusinan dan ratusan coroutine tersebut Namun, ketika jumlahnya melebihi dua ribu, saya harus memikirkan untuk mengatur antrian tugas normal dengan seorang broker, beberapa pekerja, dan sebagainya.

Pertama saya memutuskan untuk mencoba Seledri, yang pernah saya gunakan sebelumnya. Karena sifat proyek yang tidak sinkron, saya menyelami pertanyaan dan melihatnya sebuah artikeldan juga proyek, dibuat oleh penulis artikel.

Saya akan mengatakan ini, proyek ini sangat menarik dan bekerja cukup sukses di aplikasi lain dari tim kami, dan penulis sendiri mengatakan bahwa dia dapat meluncurkannya ke dalam produksi menggunakan kumpulan asinkron. Tapi, sayangnya, ternyata hal itu tidak terlalu cocok untukku masalah dengan peluncuran tugas kelompok (lihat. kelompok). Pada saat penulisan isu sudah tutup, namun pekerjaan sudah berlangsung selama sebulan. Bagaimanapun, semoga sukses untuk penulis dan semua yang terbaik, karena sudah ada hal-hal yang berfungsi di lib... secara umum, intinya ada pada saya dan alat itu ternyata lembab bagi saya. Selain itu, beberapa tugas memiliki 2-3 permintaan http ke layanan yang berbeda, jadi meskipun mengoptimalkan tugas, kami membuat 4 ribu koneksi TCP, kira-kira setiap 2 jam - tidak terlalu bagus... Saya ingin membuat sesi untuk satu jenis tugas saat memulai pekerja. Sedikit lagi tentang banyaknya permintaan melalui aiohttp di sini.

Dalam hal ini, saya mulai mencari alternatif dan menemukannya! Pencipta seledri, khususnya, menurut pemahaman saya Tanya Solem, telah dibuat Faust, awalnya untuk proyek tersebut Robin Hood. Faust terinspirasi oleh Kafka Streams dan bekerja dengan Kafka sebagai broker, rocksdb juga digunakan untuk menyimpan hasil pekerjaan agen, dan yang paling penting adalah perpustakaannya tidak sinkron.

Anda juga bisa melihatnya perbandingan cepat seledri dan faust dari pencipta yang terakhir: perbedaannya, perbedaan antara broker, pelaksanaan tugas dasar. Semuanya cukup sederhana, namun, fitur bagus di Faust menarik perhatian - data yang diketik untuk ditransfer ke topik.

Apa yang kita lakukan?

Jadi, dalam rangkaian artikel singkat, saya akan menunjukkan cara mengumpulkan data dari tugas latar belakang menggunakan Faust. Sumber untuk proyek contoh kita adalah, seperti namanya, alfavantage.co. Saya akan mendemonstrasikan cara menulis agen (sink, topik, partisi), cara melakukan eksekusi reguler (cron), perintah faust cli yang paling nyaman (pembungkus di atas klik), pengelompokan sederhana, dan pada akhirnya kami akan melampirkan datadog ( bekerja di luar kotak) dan mencoba sesuatu untuk dilihat. Untuk menyimpan data yang dikumpulkan kami akan menggunakan mongodb dan motor untuk koneksi.

PS Dilihat dari keyakinan penulisan poin tentang pemantauan, menurut saya pembaca di akhir artikel terakhir masih akan melihat seperti ini:

Persyaratan proyek

Karena apa yang telah saya janjikan, mari kita buat daftar kecil tentang apa yang seharusnya dapat dilakukan oleh layanan tersebut:

Unggah sekuritas dan ikhtisarnya (termasuk laba dan rugi, neraca, arus kas - selama setahun terakhir) - secara teratur
Unggah data historis (untuk setiap tahun perdagangan, temukan nilai ekstrim dari harga penutupan perdagangan) - secara teratur
Unggah data perdagangan terbaru - secara teratur
Unggah daftar indikator yang disesuaikan untuk setiap keamanan - secara teratur

Seperti yang diharapkan, kami memilih nama untuk proyek dari awal: Horton

Infrastruktur sedang kami persiapkan

Judulnya tentu kuat, namun yang perlu Anda lakukan hanyalah menulis konfigurasi kecil untuk docker-compose dengan kafka (dan zookeeper - dalam satu wadah), kafdrop (jika kita ingin melihat pesan dalam topik), mongodb. Kita mendapatkan [buruh pelabuhan-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) dari bentuk berikut:

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

Tidak ada yang rumit sama sekali di sini. Dua pendengar dideklarasikan untuk kafka: satu (internal) untuk digunakan di dalam jaringan komposit, dan yang kedua (eksternal) untuk permintaan dari luar, sehingga mereka meneruskannya ke luar. 2181 — pelabuhan penjaga kebun binatang. Selebihnya, menurut saya, sudah jelas.

Mempersiapkan kerangka proyek

Dalam versi dasar, struktur proyek kita akan terlihat seperti ini:

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 *

*Semua yang saya catat Kami belum menyentuhnya, kami hanya membuat file kosong.**

Kami membuat struktur. Sekarang mari tambahkan dependensi yang diperlukan, tulis konfigurasi dan sambungkan ke mongodb. Saya tidak akan memberikan teks lengkap dari file di artikel, agar tidak menundanya, tetapi saya akan memberikan tautan ke versi yang diperlukan.

Mari kita mulai dengan dependensi dan meta tentang proyek - proyek py.toml

Selanjutnya, kita mulai menginstal dependensi dan membuat virtualenv (atau Anda dapat membuat folder venv sendiri dan mengaktifkan lingkungan):

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

Sekarang mari kita berkreasi config.yml - Kredensial dan tempat untuk mengetuk. Anda dapat langsung menempatkan data untuk alphavantage di sana. Baiklah, mari kita lanjutkan ke config.py — mengekstrak data untuk aplikasi dari konfigurasi kami. Ya, saya akui, saya menggunakan lib saya - sitri.

Saat terhubung ke Mongo, semuanya cukup sederhana. diumumkan kelas klien untuk menghubungkan dan kelas dasar untuk minyak mentah, untuk memudahkan membuat kueri pada koleksi.

Apa yang akan terjadi selanjutnya?

Artikelnya tidak terlalu panjang, karena disini saya hanya berbicara tentang motivasi dan persiapan, jadi jangan salahkan saya - saya berjanji di bagian selanjutnya akan ada aksi dan grafik.

Jadi, di bagian selanjutnya ini kita:

Mari kita menulis klien kecil untuk alphavantage di aiohttp dengan permintaan untuk titik akhir yang kita butuhkan.
Mari kita buat agen yang akan mengumpulkan data sekuritas dan harga historisnya.

Kode proyek

Kode untuk bagian ini

Sumber: www.habr.com

Latar Belakang Tugas Faust, Bagian I: Pendahuluan