10년 전, 나는 우연히 Stratosphere라는 흥미로운 빅데이터 프로젝트를 알게 되었다. 내 관심을 즉시 끌었던 것은 소개 문서에 있던 한 가지 내용이었다. 단 하나의 머신 위에서 클러스터를 띄우고, MapReduce 기반의 WordCount 연산을 단지 3줄의 코드만으로 실행할 수 있다는 것이었다! 당시 하둡(Hadoop)이 주류였던 시대에 WordCount 프로그램을 설치하고 실행하는 데에는 일반적으로 몇 시간, 심지어 며칠이 걸리기도 했다. 그래서 단 3줄의 코드로 같은 기능을 달성하는 프로젝트를 만났다는 사실은 내게 강렬한 인상을 남겼다. 이러한 간결하면서도 강력한 접근 방식에 매료되어, 나는 이 프로젝트를 깊이 파고들었고, 결국 프로젝트의 컨트리뷰터가 되었다.

현재 과거 Stratosphere라 불렸던 프로젝트는 큰 변화를 거쳐, 빅데이터 분야에서 가장 인기 있는 스트림 프로세싱 엔진으로 자리잡은 Apache Flink가 되었다. 그러나 Stratosphere 초창기와는 달리, Flink는 상당히 복잡하고 거대한 프로젝트로 성장했다. 그럼에도 불구하고, Flink 스트림 프로세싱 엔진의 초기 설계와 개발에 참여했던 사람으로서, 나는 여전히 간결함을 추구하는 사용자 경험에 대한 갈망이 있다. 사용자들이 신속하게 스트림 프로세싱 여정을 시작하고, 놀라운 속도와 효율성을 체험하길 바란다.

이러한 믿음을 실현하기 위해, 나와 우리 팀은 RisingWave(https://github.com/risingwavelabs/risingwave)를 개발했다. RisingWave는 PostgreSQL과 유사한 사용 경험을 제공하며 고성능의 분산 스트림 프로세싱을 실현하는 클라우드 기반 스트리밍 데이터베이스이다. 본 기사에서는 단지 4줄의 코드만으로 RisingWave를 사용하여 스트림 프로세싱을 시작하는 방법을 소개하고자 한다.

스트림 프로세싱(Stream Processing)이란 무엇인가?

참고: 이미 스트림 프로세싱을 잘 알고 있다면 이 섹션은 건너뛰어도 좋다.

배치 처리(Batch processing)와 스트림 처리(Stream processing)는 데이터 처리의 두 가지 기본 모드이다. 지난 20년 동안 배치 처리 시스템과 스트림 처리 시스템은 각각 빠르게 진화해 왔다. 단일 머신에서 분산 시스템으로 발전하였고, 빅데이터 시대에서 클라우드 컴퓨팅 시대에 맞게 변화하였다. 특히 배치 처리 시스템과 스트림 처리 시스템 모두에서 큰 아키텍처 변화가 있었다.

The evolution of batch processing and stream processing systems over the last 20 years

배치 처리와 스트림 처리의 주요 차이점은 다음 두 가지이다.

배치 처리 시스템은 사용자가 직접 연산을 시작하는 반면, 스트림 처리 시스템은 이벤트 기반(event-driven) 연산을 수행한다.
배치 처리 시스템은 전체 데이터에 대한 계산(Full-computation model)을 수행하는 반면, 스트림 처리 시스템은 증분 계산(Incremental computation model)을 수행한다.

배치 처리든 스트림 처리든, 점점 더 실시간 처리를 추구하는 방향으로 발전하고 있다. 현재 배치 처리 시스템은 대화형 분석(interactive analysis) 시나리오에서 널리 사용되는 반면, 스트림 처리 시스템은 모니터링, 경고(alerting), 자동화 등 다양한 시나리오에서 폭넓게 사용되고 있다.

Comparison of real-time OLAP systems and real-time streaming stream processing systems

RisingWave: PostgreSQL 경험을 제공하는 스트림 프로세싱

RisingWave(https://github.com/risingwavelabs/risingwave)는 Apache 2.0 라이선스 기반의 오픈 소스 분산 SQL 스트리밍 데이터베이스이다. PostgreSQL과 호환 가능한 인터페이스를 제공하며, 사용자가 분산 스트림 프로세싱을 PostgreSQL 데이터베이스를 다루듯 쉽게 수행할 수 있도록 한다.

RisingWave는 주로 두 가지 시나리오를 위해 설계되었다: 스트리밍 ETL 및 스트리밍 분석(Streaming analytics).

스트리밍 ETL이란 다양한 데이터 소스(OLTP 데이터베이스, 메시지 큐, 파일 시스템 등)로부터 데이터를 실시간으로 가져와 조인, 집계, 그룹화, 윈도우 연산 등을 수행한 뒤, 최종 목적지 시스템(OLAP 데이터베이스, 데이터 웨어하우스, 데이터 레이크, 또는 OLTP 데이터베이스, 메시지 큐, 파일 시스템 등)으로 전달하는 작업을 말한다. 이 시나리오에서 RisingWave는 Apache Flink를 완벽히 대체할 수 있다.

반면 스트리밍 분석은 여러 데이터 소스(OLTP 데이터베이스, 메시지 큐, 파일 시스템 등)로부터 데이터를 수집하고 복잡한 분석(조인, 집계, 그룹화, 윈도우 연산 등)을 수행한 뒤, BI 대시보드에 결과를 시각화해 제공하는 것을 의미한다. 또한 사용자는 다양한 프로그래밍 언어의 클라이언트 라이브러리를 통해 RisingWave 내부의 데이터에 직접 접근할 수도 있다. 이 시나리오에서 RisingWave는 Apache Flink와 SQL/NoSQL 데이터베이스(MySQL, PostgreSQL, Cassandra, Redis 등)의 조합을 대체할 수 있다.

4줄의 코드로 RisingWave 배포하기

Mac 환경에서 RisingWave를 설치하고 실행하려면 명령줄 터미널에서 다음의 세 가지 명령어를 입력하면 된다(Linux 사용자는 다음 링크 참조: https://www.risingwave.dev/docs/current/risingwave-local/?current-os=linux):

$ brew tap risingwavelabs/risingwave
$ brew install risingwave
$ risingwave playground

다음으로, 새로운 터미널 창을 열고 다음 명령어를 실행하여 RisingWave와 연결하자:

$ psql -h localhost -p 4566 -d dev -U root

이해를 돕기 위해, 먼저 테이블을 생성하고 테스트 데이터를 추가하기 위해 INSERT를 사용해 보자. 실제 상황에서는 메시지 큐나 OLTP 데이터베이스 등에서 데이터를 가져오게 되는데, 이는 이후 설명하겠다.

웹사이트 방문 기록을 저장하기 위한 테이블을 만들어 보자:

CREATE TABLE website_visits (
  timestamp TIMESTAMP,
  user_id VARCHAR,
  page_id VARCHAR,
  action VARCHAR
);

다음으로, 각 페이지에 대한 방문 횟수, 방문자 수 및 마지막 방문 시간을 계산하는 Materialized View를 생성한다. 스트리밍 데이터를 기반으로 한 Materialized View는 RisingWave의 핵심 기능 중 하나이다.

CREATE MATERIALIZED VIEW page_visits_mv AS
SELECT page_id,
       COUNT(*) AS total_visits,
       COUNT(DISTINCT user_id) AS unique_visitors,
       MAX(timestamp) AS last_visit_time
FROM website_visits
GROUP BY page_id;

이제 INSERT를 이용해 데이터를 추가해 보자:

INSERT INTO website_visits (timestamp, user_id, page_id, action) VALUES
  ('2023-06-13T10:00:00Z', 'user1', 'page1', 'view'),
  ('2023-06-13T10:01:00Z', 'user2', 'page2', 'view'),
  ('2023-06-13T10:02:00Z', 'user3', 'page3', 'view'),
  ('2023-06-13T10:03:00Z', 'user4', 'page1', 'view'),
  ('2023-06-13T10:04:00Z', 'user5', 'page2', 'view');

현재 결과를 살펴보자:

SELECT * from page_visits_mv;

-----결과
 page_id | total_visits | unique_visitors |   last_visit_time
---------+--------------+-----------------+---------------------
 page2   |            2 |               2 | 2023-06-13 10:04:00
 page3   |            1 |               1 | 2023-06-13 10:02:00
 page1   |            2 |               2 | 2023-06-13 10:03:00
(3 rows)

데이터의 지속적 유입을 시뮬레이션하기 위해 5개 데이터를 더 추가해보자:

INSERT INTO website_visits (timestamp, user_id, page_id, action) VALUES
  ('2023-06-13T10:05:00Z', 'user1', 'page1', 'click'),
  ('2023-06-13T10:06:00Z', 'user2', 'page2', 'scroll'),
  ('2023-06-13T10:07:00Z', 'user3', 'page1', 'view'),
  ('2023-06-13T10:08:00Z', 'user4', 'page2', 'view'),
  ('2023-06-13T10:09:00Z', 'user5', 'page3', 'view');

다시 결과를 확인해 보면, 업데이트된 데이터를 볼 수 있다:

SELECT * FROM page_visits_mv;
-----결과
 page_id | total_visits | unique_visitors |   last_visit_time
---------+--------------+-----------------+---------------------
 page1   |            4 |               3 | 2023-06-13 10:07:00
 page2   |            4 |               3 | 2023-06-13 10:08:00
 page3   |            2 |               2 | 2023-06-13 10:09:00
(3 rows)

위 결과는 자동으로 최신 상태를 유지한다.

Kafka와의 연동

메시지 큐는 스트림 데이터 처리에서 일반적으로 사용되므로, Kafka에서 데이터를 실시간으로 가져와 처리하는 방법을 살펴보자.

Kafka가 설치되지 않았다면, 공식 웹사이트에서 적절한 압축 패키지를 다운로드한 후 압축을 푼다(여기서는 3.4.0 버전을 예로 들었다):

$ tar -xzf kafka_2.13-3.4.0.tgz
$ cd kafka_2.13-3.4.0

이제 Kafka를 실행해 보자.

클러스터 UUID를 생성한다:

$ KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

로그 디렉터리를 포맷한다:

$ bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties

Kafka 서버를 실행한다:

$ bin/kafka-server-start.sh config/kraft/server.properties

Kafka 서버를 실행한 후, 토픽을 생성할 수 있다:

$ bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092

Kafka 실행 후, 명령줄에서 직접 메시지를 입력할 수 있다.

먼저 Producer 프로그램을 시작하자:

$ bin/kafka-console-producer.sh --topic test --bootstrap-server localhost:9092

'>' 기호가 나타나면 JSON 형식으로 데이터를 입력한다:

{"timestamp": "2023-06-13T10:05:00Z", "user_id": "user1", "page_id": "page1", "action": "click"}
{"timestamp": "2023-06-13T10:06:00Z", "user_id": "user2", "page_id": "page2", "action": "scroll"}
{"timestamp": "2023-06-13T10:07:00Z", "user_id": "user3", "page_id": "page1", "action": "view"}
{"timestamp": "2023-06-13T10:08:00Z", "user_id": "user4", "page_id": "page2", "action": "view"}
{"timestamp": "2023-06-13T10:09:00Z", "user_id": "user5", "page_id": "page3", "action": "view"}

다음 명령어를 통해 메시지를 확인할 수 있다:

$ bin/kafka-console-consumer.sh --topic test --from-beginning --bootstrap-server localhost:9092

이제 RisingWave로 데이터를 가져와 보자. RisingWave는 이 시나리오에서 메시지 컨슈머 역할을 수행한다. psql 창에서 데이터 소스를 생성하여 Kafka와의 연결을 설정한다.

CREATE source IF NOT EXISTS website_visits_stream (
 timestamp TIMESTAMP,
 user_id VARCHAR,
 page_id VARCHAR,
 action VARCHAR
 )
WITH (
 connector='kafka',
 topic='test',
 properties.bootstrap.server='localhost:9092',
 scan.startup.mode='earliest'
 )
ROW FORMAT JSON;

(이어서 다음 파트에서 번역을 계속 진행하겠습니다.)

이어서 마지막 부분의 번역을 진행하겠습니다.

이 단계에서는 RisingWave가 데이터를 실제로 수집하여 처리를 시작할 수 있도록 Materialized View를 생성해야 한다. 이해하기 쉽게 위에서 든 예제와 유사한 형태로 Materialized View를 만든다.

CREATE MATERIALIZED VIEW visits_stream_mv AS
 SELECT page_id,
 COUNT(*) AS total_visits,
 COUNT(DISTINCT user_id) AS unique_visitors,
 MAX(timestamp) AS last_visit_time
 FROM website_visits_stream
 GROUP BY page_id;

이제 결과를 확인해 보자:

SELECT * FROM visits_stream_mv;
-----결과

 page_id | total_visits | unique_visitors |   last_visit_time
---------+--------------+-----------------+---------------------
 page1   |            1 |               1 | 2023-06-13 10:07:00
 page2   |            3 |               2 | 2023-06-13 10:08:00
 page3   |            1 |               1 | 2023-06-13 10:09:00
(3 rows)

이 단계에서 Kafka로부터 데이터를 성공적으로 받아와 처리하였다.

고급 주제: RisingWave를 이용한 실시간 모니터링 시스템 구축

실시간 모니터링(real-time monitoring)은 스트림 애플리케이션에서 매우 중요한 역할을 한다. 데이터를 실시간으로 처리하여, 결과를 즉시 시각화하여 모니터링할 수 있다. RisingWave는 데이터 소스 역할을 수행하며 Superset, Grafana 등 다양한 시각화 도구와 연결하여 실시간으로 처리된 메트릭 데이터를 보여줄 수 있다. 여러분이 직접 자신만의 스트리밍 처리 및 시각화 시스템을 구축하는 도전을 해 보길 적극 권장한다. 구체적인 단계는 사용 사례 문서에서 확인할 수 있다. 이 문서에서는 RisingWave가 시스템 성능 메트릭을 처리하고 실시간으로 Grafana에 표시하는 방법을 소개하고 있다. 본 시연은 비교적 단순하지만, 실제 데이터와 익숙한 비즈니스 환경에서 더욱 풍부하고 강력한 효과를 얻을 수 있을 것이다.

Use RisingWave for monitoring and displaying real-time results in Grafana

정리

RisingWave의 가장 큰 장점은 뛰어난 간편성에 있다. 이를 통해 사용자는 SQL을 사용하여 손쉽게 분산 스트림 처리를 수행할 수 있다. 성능 측면에서도 RisingWave는 Apache Flink와 같은 기존의 빅데이터 시대 스트림 프로세싱 플랫폼을 능가한다.

성능 향상 정도는 주목할 만하다. 특히 상태 없는(stateless) 연산에서는 약 10%~30%, 상태가 있는(stateful) 연산에서는 무려 10배 이상의 성능 향상을 보인다!

자세한 내용이 담긴 성능 보고서가 곧 발표될 예정이니 기대해 보자. 효율적인 스트림 프로세싱 플랫폼은 항상 간결성을 중시해야 한다. RisingWave는 바로 이를 제공하며 더 많은 가능성을 열어준다.

RisingWave 소개

RisingWave는 클라우드 환경에 최적화된 분산 SQL 스트리밍 데이터베이스이다. Apache 2.0 라이선스 기반의 오픈소스 시스템이다.

공식 웹사이트: risingwave.com
GitHub: risingwave.com/github
공식 문서: risingwave.dev

This article was originally published on https://risingwave.com/blog/start-your-stream-processing-journey-with-just-4-lines-of-code/

Flink 너무 복잡하신가요? 단 4줄 코드로 시작하는 실시간 스트림 처리 입문 가이드