파우스트에 대한 배경 작업, XNUMX부: 소개

내가 어쩌다가 이렇게 살게 됐지?

얼마 전 저는 로드가 많은 프로젝트의 백엔드 작업을 해야 했습니다. 이 프로젝트에서는 복잡한 계산과 타사 서비스 요청이 포함된 수많은 백그라운드 작업의 정기적인 실행을 구성해야 했습니다. 이 프로젝트는 비동기식이며 제가 오기 전에는 크론 실행 작업을 위한 간단한 메커니즘이 있었습니다. 즉, 현재 시간을 확인하는 루프와 수집을 통해 코루틴 그룹을 시작하는 것이었습니다. 이러한 접근 방식은 그러한 코루틴이 수십, 수백 개가 있을 때까지 허용 가능한 것으로 나타났습니다. 그러나 그 수가 XNUMX명을 초과하면 브로커, 여러 작업자 등으로 일반 작업 대기열을 구성하는 것에 대해 생각해야 했습니다.

먼저 이전에 사용했던 셀러리를 사용해보기로 결정했습니다. 프로젝트의 비동기적 특성으로 인해 저는 질문에 깊이 빠져들어 보았습니다. статью뿐만 아니라 계획, 기사 작성자가 작성했습니다.

이 프로젝트는 매우 흥미롭고 우리 팀의 다른 애플리케이션에서 매우 성공적으로 작동하며 저자 자신도 비동기식 풀을 사용하여 이 프로젝트를 프로덕션에 출시할 수 있었다고 말합니다. 하지만 불행히도 그것은 나에게 별로 적합하지 않았습니다. 문제 그룹 작업 시작(참조: 그룹). 글을 쓰는 시점에서 발행물 이미 문을 닫았지만 작업은 한 달 동안 진행되었습니다. 어쨌든 작성자에게 행운이 있기를 바랍니다. 이미 lib에 작업 중인 작업이 있기 때문입니다. 일반적으로 요점은 나에게 있고 도구는 나에게 습한 것으로 판명되었습니다. 또한 일부 작업에는 다양한 서비스에 대한 2-3개의 http 요청이 있었기 때문에 작업을 최적화하더라도 약 4시간마다 2개의 TCP 연결을 생성합니다. 그다지 좋지는 않습니다... 한 가지 유형의 세션을 만들고 싶습니다. 작업자를 시작할 때 작업. aiohttp를 통한 대량 요청에 대해 좀 더 자세히 알아보세요. 여기에.

이와 관련하여 나는 찾기 시작했습니다. 대안 그리고 찾았어요! 특히 내가 이해하는 바에 따르면 셀러리의 창시자는 솔렘에게 물어보세요, 생성되었습니다 파우스트, 원래 프로젝트용 robinhood. Faust는 Kafka Streams에서 영감을 받아 Kafka를 브로커로 사용하고, RocksDB는 에이전트 작업 결과를 저장하는 데에도 사용되며, 가장 중요한 것은 라이브러리가 비동기식이라는 것입니다.

또한, 당신은 볼 수 있습니다 빠른 비교 후자 제작자의 셀러리와 파우스트 : 차이점, 브로커 간의 차이점, 기본 작업 구현. 모든 것이 매우 간단하지만 파우스트의 멋진 기능이 관심을 끌고 있습니다. 즉, 주제로 전송하기 위해 입력된 데이터입니다.

우리가 뭘 할거 니?

그래서 짧은 기사 시리즈에서 Faust를 사용하여 백그라운드 작업에서 데이터를 수집하는 방법을 보여 드리겠습니다. 예제 프로젝트의 소스는 이름에서 알 수 있듯이 다음과 같습니다. alphavantage.co. 에이전트 작성 방법(싱크, 토픽, 파티션), 일반(cron) 실행 방법, 가장 편리한 faust cli 명령(클릭에 대한 래퍼), 간단한 클러스터링을 시연하고 마지막에는 datadog( 즉시 작업) 무언가를 보려고 노력하십시오. 수집된 데이터를 저장하기 위해 mongodb와 모터를 사용하여 연결합니다.

추신: 모니터링에 대한 내용이 자신감있게 쓰여진 것으로 판단하면, 마지막 기사 끝 부분의 독자는 여전히 다음과 같이 보일 것이라고 생각합니다.

프로젝트 요구사항

이미 약속했으므로 서비스가 수행할 수 있는 작업에 대한 간단한 목록을 만들어 보겠습니다.

1. 증권 및 개요(지난해 손익, 대차대조표, 현금 흐름 포함)를 정기적으로 업로드하세요.
2. 과거 데이터 업로드(매 거래 연도마다 거래 종가의 극단값 찾기) - 정기적으로
3. 최신 거래 데이터 업로드 - 정기적으로
4. 각 증권에 대한 맞춤형 지표 목록을 정기적으로 업로드

예상대로 처음부터 프로젝트 이름을 선택합니다. 호튼

인프라를 준비하고 있습니다

제목은 확실히 강력하지만 kafka(및 Zookeeper - 하나의 컨테이너에 있음), kafdrop(주제의 메시지를 보려는 경우), mongodb를 사용하여 docker-compose에 대한 작은 구성을 작성하기만 하면 됩니다. 우리는 얻는다 [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

여기에는 전혀 복잡한 것이 없습니다. Kafka에 대해 두 개의 리스너가 선언되었습니다. 하나(내부)는 복합 네트워크 내부에서 사용하기 위한 것이고, 두 번째(외부)는 외부 요청을 위한 것이므로 외부로 전달했습니다. 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

다음으로 종속성을 설치하고 virtualenv를 생성하기 시작합니다(또는 venv 폴더를 직접 생성하고 환경을 활성화할 수도 있음).

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

이제 만들어 봅시다 config.yml - 자격 증명 및 노크 장소. 거기에 Alphavantage용 데이터를 즉시 배치할 수 있습니다. 글쎄, 다음으로 넘어가자 config.py — 구성에서 애플리케이션에 대한 데이터를 추출합니다. 네, 고백합니다. 저는 lib를 사용했습니다. 시트리.

Mongo에 연결하면 모든 것이 매우 간단합니다. 발표 클라이언트 클래스 연결하고 기본 클래스 크루드의 경우 컬렉션에 대한 쿼리를 더 쉽게 만들 수 있습니다.

그 다음은 어떻게됩니까?

여기서는 동기 부여와 준비에 대해서만 이야기하고 있으므로 기사가 그리 길지 않으므로 저를 비난하지 마십시오. 다음 부분에는 액션과 그래픽이 있을 것이라고 약속합니다.

따라서 바로 다음 부분에서는 다음을 수행합니다.

1. 필요한 엔드포인트에 대한 요청을 사용하여 aiohttp에 alphavantage용 소규모 클라이언트를 작성해 보겠습니다.
2. 증권 및 과거 가격에 대한 데이터를 수집하는 에이전트를 만들어 보겠습니다.

프로젝트 코드

이 부분의 코드

출처 : habr.com