스트림 처리 엔진 - Arroyo 소개와 기본 사용법

소개

핵심 기능

• SQL로 정의된 파이프라인, 복잡한 분석 쿼리 지원

• 초당 수백만 개의 이벤트까지 확장 가능

• 윈도우 및 조인과 같은 상태 저장 작업

• 내결함성 및 파이프라인 복구를 위한 상태 저장

• 워터마크 지원을 통한 이벤트 시간 처리

사용 사례

• 사기 및 보안 사고 탐지

• 실시간 제품 및 비즈니스 분석

• 데이터 웨어하우스 또는 데이터 레이크로 실시간 수집

• 실시간 ML 피처 생성

왜 Arroyo 인가?

• 서버리스 운영: Arroyo 파이프라인은 최신 클라우드 환경에서 실행되도록 설계되어 원활한 확장, 복구 및 일정 재조정을 지원합니다.

• 고성능 SQL: SQL은 일관되게 우수한 성능으로 최우선 순위의 관심사입니다.

• 비전문가를 위한 설계: Arroyo는 파이프라인 API를 내부 구현에서 깔끔하게 분리합니다. 스트리밍 전문가가 아니어도 실시간 데이터 파이프라인을 구축할 수 있습니다.

Arroyo와 Apache Flink 비교

• 유연성: Flink는 복잡한 이벤트 처리와 다양한 처리 모델을 지원합니다. Arroyo는 Flink와 같은 수준의 유연성을 목표로 하지 않습니다.

• 슬라이딩 윈도우 쿼리: Flink에서는 슬라이드를 줄일수록 처리량이 급격히 감소하는 반면, Arroyo는 슬라이드와 윈도우 크기에 상관 없이 일정한 처리량을 제공합니다.

• 체크포인트 저장소: Flink는 이벤트 상태 체크포인트를 RocksDB에 저장합니다. Arroyo는 S3와 같은 오브젝트 스토리지에 저장합니다.

• 사용 사례: Flink는 복잡한 데이터 파이프라인 구축, 실시간 분석, 이벤트 기반 애플리케이션 개발 등 다양한 고급 사용 사례에 적합합니다. Arroyo는 비전문가도 SQL 쿼리를 사용하여 실시간 스트리밍 데이터를 변환하거나 분석 할 수 있습니다.

• 개발 언어: Flink는 Java와 Scala로 개발되었으며, Arroyo는 Rust로 개발되었습니다.

아키텍처

• Web UI: React로 작성된 단일 페이지 앱으로, REST API를 통해 시스템과 상호작용합니다.

• Arroyo API: Web UI를 지원하며, REST API를 통해 모든 구성 작업과 파이프라인 관리를 처리합니다.

• Arroyo Controller: 데이터베이스에 정의된 시스템의 원하는 상태와 실제 상태를 지속적으로 조정합니다.

• Schedulers: 다양한 스케줄러를 지원하며, 데이터플로우 엔진을 구성하는 Worker를 관리 합니다.

• Arroyo Worker: 실제 처리 로직을 실행하는 워커로, 데이터플레인을 구성합니다.

• Postgres: 설정 및 시스템 상태를 저장하는 데 사용됩니다.

• Prometheus: 메트릭 수집에 사용됩니다.

• S3: 체크포인트를 저장하는 데 사용됩니다.

설치

로컬 설치 및 실행

docker run -p 8000:8000 ghcr.io/arroyosystems/arroyo-single:latest

쿠버네티스 클러스터에 배포

• arroyo-compiler

• arroyo-controller

• arroyo-api

• prometheus

• postgres

helm repo add arroyo https://arroyosystems.github.io/helm-repo

• values.yaml

artifactUrl: "s3://<YOUR_ARROYO_ARTIFACT_BUCKET>"
checkpointUrl: "s3://<YOUR_ARROYO_CHECKPOINT_BUCKET>"

• values.yaml

postgresql:
  deploy: false
  externalDatabase:
    host: postgresql.arroyo.svc.cluster.local
    name: arroyo_test
    user: arroyodb
    password: arroyodb
prometheus:
  deploy: false
  endpoint: prometheus.arroyo.svc.cluster.local
artifactUrl: "s3://<YOUR_ARROYO_ARTIFACT_BUCKET>"
checkpointUrl: "s3://<YOUR_ARROYO_CHECKPOINT_BUCKET>"

helm install arroyo arroyo/arroyo -f values.yaml

$ kubectl get pods
NAME                                        READY   STATUS             RESTARTS      AGE
arroyo-compiler-ccd6b7bdb-752vt             1/1     Running            0             36s
arroyo-controller-75587f886b-k9drg          1/1     Running            1 (18s ago)   36s
arroyo-postgresql-0                         1/1     Running            0             26s
arroyo-api-5dccb89967-zl727                 1/1     Running            2 (17s ago)   36s
arroyo-prometheus-server-5c8d49b85d-xwl2h   2/2     Running            0             36s

kubectl port-forward service/arroyo-api 8000:80

기본 사용법

연결 생성

• Connections → Create Connection→ Nexmark 카드 메뉴의 Create 클릭

• Event rate (messages / sec): 100 → Next 클릭

• Connection Name: nexmark → Test Connection 클릭 → Create 클릭

스트리밍 파이프라인 만들기

SELECT bid FROM nexmark WHERE bid IS NOT NULL;

• → Start Preview 클릭

SELECT avg(bid.price) as avg_price
FROM nexmark
WHERE bid IS NOT NULL
GROUP BY hop(interval '2 seconds', interval '10 seconds');

SELECT * FROM (
    SELECT *, ROW_NUMBER() OVER (
        PARTITION BY window
        ORDER BY count DESC) AS row_num
    FROM (SELECT count(*) AS count, bid.auction AS auction,
        hop(interval '2 seconds', interval '60 seconds') AS window
            FROM nexmark WHERE bid is not null
            GROUP BY 2, window)) WHERE row_num <= 5

참고자료

• https://doc.arroyo.dev/

• https://www.arroyo.dev/blog/how-arroyo-beats-flink-at-sliding-windows

• https://www.arroyo.dev/blog/why-not-flink