Azure-EventHub

EventHub는 짧은 대기 시간으로 초당 수백만개의 이벤트를 스트리밍할 수 있는 클라우드의 네이티브 데이터 스트리밍 서비스이다.

• Azure Stream Analytics를 사용하여 실시간 인사이트를 생성하여 이벤트 허브에서 데이터를 처리
• Azure Data Explorer를 사용하여 스트리밍 데이터를 분석하고 탐색.

Azure Event Hubs의 Apache Kafka

• AMQP(고급 메시지 큐 프롵토콜), Apache Kafka 및 HTTPS 프로토콜을 기본적으로 지원하는 다중 프로토콜 이벤트 스트리밍 엔진

Apache Kafka 용 Azure Event Hubs

Event Hubs의 스키마 레지스트리

• Event Hubs의 Azure 스키마 레지스트리는 이벤트 스트리밍 애플리케이션의 스키마를 관리하기 위한 중앙 집중식 리포지토리를 제공
  
• 스키마 레지스트리는 이벤트 생산자 및 소비자 간에 데이터 호환성 및 일관성을 보장.
• 스키마 레지스트리는 기존 Kafka 애플리케이션과 통합되며 Avro 및 JSON 를 비롯한 여러 스키마 형식을 지원한다

Reference

Event Hub의 Azure 스키마 레지스트리

Stream Analytics를 사용하여 스트리밍 이벤트 실시간 처리

• Azure Steam Analytics와 통합하여 실시간 스트림 처리를 지원
• SQL 기반 Stream Analytics 쿼리 언어를 사용하여 실시간 스트림 처리를 수행하고 스트리밍 데이터를 분석하기 위한 다양한 함수를 활용할 수 있다.

Azure Stream Analytics 통합 섹션

Azure Data Explorer를 사용하여 스트리밍 데이터 탐색

• 거의 실시간으로 대량의 데이터를 분석할 수 있는 빅 데이터 분석을 위한 완전 관리형 플랫폼
• Azure Data Explorer와 통합하면 스트리밍 데이터의 거의 실시간 분석 및 탐색 수행 가능

Event Hub에 데이터 연결

작동 방식

Event Hubs는 이벤트 소비자와 이벤트 생산자를 분리하는 시간 보존 버퍼가 있는 통합 이벤트 스트리밍 플랫폼을 제공

• Event Hubs의 주요 기능 구성 요소
  • 생산자 애플리케이션: Event Hubs SDK 또는 Kafka 생산자 클라이언트를 사용하여 이벤트 허브에 데이터를 수집.
  • 네임스페이스: 하나 이상의 이벤트 허브 또는 Kafka 토픽에 대한 관리 컨테이너.
    스트리밍 용량 할당, 네트워크 보안 구성, 지역 재해 복구 사용과 같은 관리 작업이 네임스페이스 수준에서 처리.
  • Event Hubs/Kafka 토픽: Event Hubs에서 이벤트를 이벤트 허브 또는 Kafka 토픽으로 구성할 수 있다. 이는 하나 이상의 파티션을 구성할 수 있는 추가 전용 분산 로그이다.
  • 파티션: 이벤트 허브의 크기를 조정하는 데 사용. 파티션은 고속도로의 차선 같다. 스트리밍 처리량이 더 필요한 경우 파티션을 더 추가할 수 있다.
  • 소비자 애플리케이션: 이 애플리케이션은 이벤트 로그를 검색하고 소비자 오프셋을 유지 관리하여 데이터를 사용할 수 있다. 소비자는 Kafka 소비자 클라이언트 또는 Event Hubs SDK 클라이언트일 수 있다.
  • 소비자 그룹: 이 논리적 소비자 인스턴스 그룹은 이벤트 허브 또는 Kafka 토픽에서 데이터를 읽는다. 이를 통해 여러 소비자가 각자의 속도와 자체 오프셋을 사용하여 이벤트 허브에서 동일한 스트리밍 데이터를 독립적으로 읽을 수 있다.

장점

• 확장성: 대량의 메시지를 처리할 수 있어 고부하 시스템에 적합.
• 내구성: 메시지를 안정적으로 저장하고, 소비자가 언제든 가져갈 수 있음.
• 분산 처리: 메시지 파티션을 활용해 다중 소비자가 병렬로 처리 가능.
• 통합 가능성: Azure 생태계(예: Azure Functions, Logic Apps)와 손쉽게 통합.

단점

• 실시간성: WebSocket이나 gRPC에 비해 레이턴시가 클 수 있음.
• 양방향 통신 부족: 주로 Publisher-Subscriber 모델로 동작하며, 클라이언트가 서버로 메시지를 보낼 때 직접적인 스트리밍 채널을 제공하지 않음.
• 초기 설정 복잡성: Event Hub 설정 및 인증 관리가 비교적 복잡.

적합한 사용 사례

• 높은 부하(대량의 사용자 또는 메시지)를 처리해야 하며, 메시지 손실을 최소화해야 하는 시스템.
• 비동기적으로 LLM 응답을 처리하는 애플리케이션.
• 데이터 분석 또는 아카이빙이 필요한 서비스

Azure Event Hubs: 네이티브 Apache Kafka 지원을 사용하는 실시간 데이터 스트리밍 플랫폼

Azure Eventhub에 있는 정보를 클라이언트가 바로 가져가는 것도 가능한가?

Azure Event Hubs에서 정보를 클라이언트가 바로 가져가는 방식은 기본적으로 Event Hubs의 동작 방식과 설계에 따라 제한적.

Event Hubs는 데이터 수집 및 분산 처리를 위해 설계된 이벤트 스트림 플랫폼으로, 직접 접근보다는 소비자 그룹(Consumer Group)을 통해 데이터를 처리하는 것이 일반적이다.

직접 클라이언트 접근 가능 여부

기본적으로 불가능

• Event Hubs는 메시지 브로커 역할을 하며, 데이터를 생산자(Producer)와 소비자(Consumer) 간에 비동기적으로 전달한다.
• 클라이언트가 Event Hubs에 직접 접근해서 데이터를 가져가도록 설계되어 있지 않다.
  • Event Hubs는 데이터를 Offset 및 Partition 단위로 관리하여 소비자가 순서에 따라 데이터를 처리하도록 설계됨.
  • 클라이언트가 바로 가져가려면 Event Hubs의 읽기 API를 구현해야 하고, 이 과정에서 메시지 순서나 Offset 관리가 필요.

---

가능한 대안 및 설계 방안

1. Azure Functions or Logic Apps를 사용한 중계

• Event Hubs의 데이터를 트리거로 읽고, 이를 클라이언트에게 전달하는 REST API 또는 WebSocket 인터페이스를 만든다.
• 클라이언트는 Event Hubs 대신 중계 서버를 통해 데이터를 요청.

2. Azure Stream Analytics와 실시간 데이터 제공

• Event Hubs 데이터를 실시간으로 처리하여 결과를 클라이언트가 사용할 수 있도록 데이터베이스나 다른 서비스에 저장.
• 예: 처리된 데이터를 Azure Cosmos DB, SQL Database 또는 Blob Storage에 저장.

3. Azure SignalR Service와 WebSocket 통합

• Event Hubs 데이터를 소비자가 실시간으로 처리한 뒤, SignalR을 사용해 데이터를 클라이언트로 푸시.
• SignalR은 클라이언트와 지속적인 WebSocket 연결을 유지하여 실시간 데이터 전송을 가능하게 한다.

4. Event Processor Host (SDK 기반 소비)

• 클라이언트가 직접 Event Hubs의 데이터를 가져가려면 Event Hubs SDK를 활용하여 Consumer로 동작.
• 이 경우, 클라이언트는 Event Hubs에 Consumer Group으로 등록되고, 데이터를 처리해야 한다.
• 하지만 이는 클라이언트가 직접 Offset 관리와 같은 복잡성을 처리해야 하므로 일반적으로 권장되지 않음.

---

추천 아키텍처

1. Event Hubs → Azure Functions → REST API

• Event Hubs 데이터를 Azure Functions로 처리하고, REST API로 제공.
• 클라이언트는 REST API를 통해 데이터를 요청.

2. Event Hubs → Stream Analytics → Cosmos DB → 클라이언트

• Stream Analytics로 데이터를 처리한 뒤, 클라이언트가 저장된 데이터를 DB에서 조회.

3. Event Hubs → SignalR Service → WebSocket 클라이언트

• 실시간 데이터 제공이 필요한 경우 SignalR을 사용해 WebSocket으로 전송.

---

결론

Event Hubs는 클라이언트가 직접 접근해서 데이터를 가져가도록 설계되지 않았다.

클라이언트가 데이터를 직접 요청하거나 실시간으로 수신해야 하는 경우, 중간 계층을 설계하는 것이 최적의 방법이다.
중계 서버나 다른 Azure 서비스를 활용하여 데이터를 안정적으로 제공하도록 아키텍처를 구성하는 것을 권장한다.

설계 개요

• 채팅 응답 발행 (Producer) → Agent가 Event Hub로 응답 전송
• 채팅 응답 수신 (Consumer 서버) → Event Hub에서 메시지 수신 후 클라이언트로 HTTP 요청 전송
• 클라이언트로 응답 전달 → HTTP API

주요 기술 스택

역할	기술	설명
Producer	FastAPI	Agent 응답을 Event Hub에 발행
Event Hub	Azure Event Hub	메시지 브로커 역할
Consumer	Legacy 서버	Event Hub에서 메시지 읽기
클라이언트 응답	HTTP API	클라이언트에 직접 전송

상세 흐름 (예시)

채팅 응답 발행 (Producer)

from azure.eventhub import EventHubProducerClient, EventData

# EVENT_HUB_CONNECTION_STRING : Azure Event Hub 인스턴스에 연결할 때 필요한 인증 정보
producer = EventHubProducerClient.from_connection_string(
    "EVENT_HUB_CONNECTION_STRING", eventhub_name="chat-responses"
)

event_data = EventData('{"user_id": "123", "message": "Hello!"}')
producer.send_batch([event_data])
producer.close()

채팅 응답 수신 (Consumer, Legacy 서버)

서버에서 Event Hub 메시지를 읽고 즉시 클라이언트로 HTTP 요청 전송

import asyncio
import aiohttp
from azure.eventhub.aio import EventHubConsumerClient
import json

async def send_to_client(user_id, message):
    async with aiohttp.ClientSession() as session:
        async with session.post(f"http://client-server.com/chat/{user_id}", json={"message": message}) as resp:
            print(f"Sent to client {user_id}, status: {resp.status}")

async def process_event(event):
    data = json.loads(event.body_as_str())
    user_id = data["user_id"]
    message = data["message"]

    # HTTP API로 클라이언트에 전송
    await send_to_client(user_id, message)

async def on_event(partition_context, event):
    await process_event(event)
    await partition_context.update_checkpoint(event)

async def main():
    client = EventHubConsumerClient.from_connection_string(
        "EVENT_HUB_CONNECTION_STRING", consumer_group="$Default", eventhub_name="chat-responses"
    )
    async with client:
        await client.receive(on_event=on_event, starting_position="-1")

asyncio.run(main())

• user_id 또는 transaction_ID 등을 이용하여 클라이언트 구분 로직 필요

특징

장점

• 구현이 단순: Redis와 WebSocket이 필요 없음
• 실시간 전송 가능: 서버에서 바로 클라이언트로 HTTP 요청 전송
• 확장성 높음: 여러 Consumer 서버가 메시지를 처리 가능

단점

• 클라이언트가 항상 온라인이어야 함 → 클라이언트가 오프라인이면 메시지 손실 가능

결론

• 클라이언트가 항상 온라인 상태라면 WebSocket 없이도 구현 가능
• 클라이언트가 메시지를 놓치지 않으려면 Redis나 DB로 클라이언트 별 메시지를 저장하고 있어야 함 (Legacy 서버)

출처

1. Azure Event Hubs

• 📄 공식 문서:
  • Azure Event Hubs 개요
  • Event Hubs Python SDK 사용법
• 📦 GitHub (Azure SDK for Python)
  • Azure SDK for Python (eventhub)

---

2. Python을 사용한 Event Hub 소비자 (Consumer) 구현

• Azure Event Hub에서 메시지 소비 (Python)
• Azure Event Hub Consumer Client

---

3. 기타 Azure 관련 문서 (배포 및 운영)

• Azure Monitor 및 Application Insights

• Azure Kubernetes Service(AKS)로 Event Hub 배포

Azure Event Hub Pub/Sub 하기 위한 인증 방법

이벤트허브 만들기