Elastic Search(OpenSearch) 장애 사례

장애 내용

OpenSearch의 샤드 배치가 안됐다.

장애 원인

사실 주니어 개발자였고, ES 관련된 부분은 많이 찾아보지 않아서 여러모로 확인에 애를 먹고 있었다. 사용하는것과 관리하는것은 다른 부분이기에 그러한 설정 부분과 공식문서 등을 유심히 찾아본 결과 클러스터에 '각 노드에 설정 가능한 최대 샤드 갯수'를 설정할 수 있다는 사실을 알게 되었다.

 cluster.max_shards_per_node 

해당 옵션의 default 는 1000개인데, 알고보니 개발환경 OpenSearch는 이미 1000개가 배치되어 있었다.

장애 해결 방법

해결 방안은 두 가지다

• 가장 쉬운 방법 - OpenSearch 클러스터에 노드를 추가한다. (비용 증가, 안정성 확보)
• 옵션 조정을 통해 노드당 샤드 최대 갯수를 증가시킨다. (비용 동결, 안정성 하락)

현업에서는 모든 부분에 있어 다양한 요소를 두고 판단한다. 당연히 계속 확장하면 좋지만 그 비용도 우리가 내야한다. 특히 상황이 어려울수록 클라우드 비용 절감은 중요한 요소이다 그러나 우리는 회의 끝에, 노드를 증가시키는 의사 결정을 했고, 이를 통해 안정성을 확보했다.

느낀점

단순히 ES의 사용 뿐 아니라 인프라 관리 및 설정 등에 대해서도 중요성을 체감할 수 있었다.

필자는 Jupyter Notebook, pyspark, great expectation을 사용하였다. dataset은 https://github.com/databricks/LearningSparkV2/tree/master/databricks-datasets/learning-spark-v2/flights 의 departuredelays.csv를 사용하였다.

1. import

from pyspark.sql import SparkSession
import great_expectations as gx

2. 일단 SparkSession 만들고 dataframe 생성

spark = SparkSession.builder.appName("test_app").getOrCreate()
df = spark.read.option("header", "true").option("inferSchema", "true").csv("departuredelays.csv")
# inferSchema 옵션을 true로 할 경우 처리 시간이 오래 걸린다. 현업에서는 스키마 따로 정의하자

---
> 3. great expectaion 시작. data context 생성
```python
context = gx.get_context(mode="ephemeral")
# 간단한 테스트니까 ephemeral 도 충분하다.

4. datasource 생성

data_source = context.data_sources.add_spark(name="test_source")
# source는 저번에 말했듯 datalake 같은 저장소에 대한 것이다

5. data asset 생성

data_asset = data_source.add_dataframe_asset(name="test_asset")
# asset은 곧 dataset 같은 단위라고 보면 된다.

6. batch_definition 생성

batch_def = data_asset.add_batch_definition_whole_dataframe("test_batch")
# batch는 위의 data asset, 즉 dataset이 실제로는 매우 클 때가 있는데 (수백 Gb ~ Tb 단위)
# 이를 한 번에 처리하기 어려우므로 파티셔닝해서 처리할 때 쓰인다.
# 우리는 예시이므로 asset 전체를 batch로 가져가는 add_batch_definition_whole_dataframe 를 사용했다.

7. expectation 생성

expectations_distance =  gx.expectations.ExpectColumnValuesToBeBetween(column="distance", max_value=5000, min_value=10)
# 여기서부터는 dataset에 따라, 조직의 요구사항에 따라 dataset quality check 요소를 expectation으로 만든다.
# 나는 항공기의 출발지,도착지,거리, 지연 등을 담은 데이터셋이므로 거리에 대해 체크했다.

8. expectation suite 생성 및 expectation 추가

expectation_suite_ref = gx.ExpectationSuite(name="test_suite")
expectation_suite = context.suites.add(expectation_suite_ref)
expectation_suite.add_expectation(expectations_distance)
# expectation_suite를 먼저 만들어주고, 이 후에 expectation을 suite에 추가한다.

9. validation 생성

validation = gx.ValidationDefinition(data=batch_def, suite=expectation_suite, name="test_valid")
# 이제 실제로 수행할 validation에 우리의 작업 단위와 평가 기준 등을 담아준다.

10. validation 수행

batch_params = {"dataframe": df}
result = validation.run(batch_parameters=batch_params)
# dataframe을 사용한다고 정의해주고, 이를 실행해서 결과를 얻는다.

11. result 확인

print(result)
# 확인 결과는 아래와 같다.
{
  "success": true,
  "results": [
    {
      "success": true,
      "expectation_config": {
        "type": "expect_column_values_to_be_between",
        "kwargs": {
          "batch_id": "test_source-test_asset",
          "column": "distance",
          "min_value": 10.0,
          "max_value": 5000.0
        },
        "meta": {},
        "id": "b692dcb0-1813-4e82-91ce-8c045572d107",
        "severity": "critical"
      },
      "result": {
        "element_count": 1391578,
        "unexpected_count": 0,
        .
        .
        .
#성공적으로 수행되었음을 확인했다. 아래의 결과중 중요한 부분은
# success: expectation을 만족하지 못하면 false
# 그 이후 우리가 정의한 expectation type을 확인하고, element_count와 unexpected_count를 통해
# 실패케이스를 확인하고 카운팅하고 한다.

실패 케이스도 쉽게 만들 수 있으나, 여러분들이 해보시길 권장한다. 이렇게 가볍게 spark와 great_expectation을 연동하여 test를 해봤다.

공식 깃허브, 대략 10.9k star면 훌륭한 것 같다.

gx-core와 gx cloud가 존재한다. gx-core의 경우 Apache-2.0 license 이다.

기본적으로는 python 기반 품질 관리 도구이고, 다양한 data engineering 도구들과 연동할 수 있다. 필자는 spark를 주로 사용하므로 이에 대해 같이 사용해 볼 것이다.

보통은 datalake에서 그 이후 warehouse, mart 및 ML/DL 모델 등 다양한 곳으로 데이터가 흘러갈텐데, 이 때 이 흘러갈 data의 quality를 검증한다고 생각하면 된다.

기본적인 구조

• Data Context: 프로젝트의 전체 설정과 메타데이터 관리. 3가지 타입 존재
  • Ephemeral: In-Memory, 메타데이터를 영속적으로 저장하지는 않음. PoC 용도
  • FilesystemData: 가장 기본적인 형태. 메타데이터를 filesystem에 저장
  • Cloud: 위에서 언급한 gx cloud를 사용할 때 사용 가능
• Data Source: 데이터가 저장되어 있는곳. Databricks, RDBMS, Spark, Pandas 등
• Data Asset: Source에 저장된 dataset을 의미한다고 보면 됨. 즉 우리의 quality testing 대상
• Batch Definition: Asset을 이제 어떻게 처리할것인가에 대한 설정. batch size 등
• Expectation Suite: 우리가 정의할 Expectations(데이터 품질 규칙) 집합
• Expectation: 데이터에 대한 품질 규칙. 예를 들어
  • col1은 not null 이어야 함
  • col2는 정수값인데 max=100, min=0 범위 안에 있어야 함. 등
• Validation: 위의 Batch Definition과 Excetation Suite를 사용해서, 즉 정의된 batch 대상에 해당 suite 적용

3편에서 실제 spark와 gx-core의 연동을 다뤄보겠다.

데이터 품질

Data Quality가 중요한 이유

• Data를 대시보드 등을 구성하고 의사결정에 사용하는데 데이터가 잘못됐다면?
• Data Quality의 이슈로 인해 ML/DL 모델 학습에 문제가 생긴다면?

이런 이슈를 Data Quality Management를 통해 방지하고자 함

Data Quality 속성

• 정확성 (Accuracy)

  • 데이터에서 잘못된 값, 오타, 왜곡된 정보와 같은 오류를 제거
  • 정확성을 측정하고 개선하기 위한 방법
    • 검증 절차(Validation checks)
    • 신뢰할 수 있는 출처로 교차 검증(Cross-referencing with trusted sources)
    • 정기적인 감사(Regular audits)

• 완전성 (Completeness)

  • 데이터셋에 모든 필수 데이터가 존재하고 사용 가능한 정도
  • 모든 필요한 속성이 완전히 채워져 있어야 함
  • 필수 속성 (Required attributes)
    • 데이터셋이 완전하다고 간주되려면 반드시 값이 입력되어야 함
  • 선택 속성 (Optional attributes)
    • 필수는 아니지만 추가적인 인사이트를 제공

• 일관성 (Consistency)

  • 데이터의 일관성. 이를 보장하기 위해 아래와 같은 방법들 활용
  • 동기화 프로세스 (Synchronization processes)
    • 한 시스템에서 이루어진 업데이트가 다른 모든 시스템에도 정확히 반영
  • 검증 규칙 (Validation rules)
    • 시스템 간 불일치(discrepancy)를 감지하고 해결할 수 있도록 지원합니다.

• 적시성 (Timeliness)

  • 데이터가 필요한 시점에 이용 가능하고 최신 상태인지에 대한 척도
    • 적시에 데이터를 제공함으로써 가장 최신·관련성 높은 정보를 기반으로 의사결정 가능

• 유효성 (Validity)

  • 데이터셋의 값이 정의된 규칙, 표준 및 기대치에 부합하는 정도
  • 데이터가 사전에 정해진 기준(타입, 형식,범위, 제약 조건 등)을 준수하도록 보장

• 고유성 (Uniqueness)

  • 각 레코드가 중복되지 않고 고유함
  • 이를 보장하기 위해 다음과 같은 방법을 사용
    • 엄격한 검증 규칙 적용(Strict validation rules)
    • 중복 제거 프로세스(Deduplication processes) 활용

Data Quality 관련 도구

• Great Expectations (가장 유명한 도구)
• Soda (좀 더 sql 친화적?이라고 알고있음)
• Deequ (아마존이 개발함)

다음편에서는 Great Expectaion(중에서 gx-core)의 기본 개념 및 구성 등에 대해 다루겠음

• Compact and Aligned Text API

가장 많이 사용되는 API 4개

• _cat/health
  • 클러스터의 전반적인 상태확인 API
  • 상태 종류
    • green : 프라이머리, 레플리캬 사드 모두 정상적으로 배치
    • yellow : 프라이머리 정상, 레플리카 비정상 -> 검색 성능 악영향
    • red : 프라이머리와 레플리카 둘 다 비정상
• _cat/nodes
  • 기본적인 노드 정보 확인 API
  • h 옵션을 통해 header 지정 -> 원하는 정보 filtering
• _cat/indices
  • 인덱스 상태 확인 API
  • 인덱스 프라이머리 샤드 & 레플리카 샤드 갯수 확인
  • 이상 상태 인덱스 확인
• _cat/shards
  • 샤드 상태 확인

?v 하면 사람이 볼 수 있는 형태(verbose)

Elastic Search 주요 지표

** 모니터링 도구 선정 **

• AWS OpenSearch : AWS CloudWatch
• ElasticSearch : Kibana.. 그 외 promtheus도 고려할 만 함

어떤 지표를 모니터링 할 것인가

• ** 알람을 받아서 처리해야 할 지표 **
  • CPU Usage : 노드가 CPU를 얼마나 많이 사용하고 있는가 - 50% 이상
  • Disk Usage : 노드가 얼마나 많은 문서를 저장하고 있는가 - 70% 이상
  • Load : 노드가 얼마나 많은 CPU 및 Disk 연산을 처리하는가 (부하) - cpu 갯수 따라서
  • JVM Heap : JVM 메모리 사용량 - 꺾이는 등 정상적인 GC 상태를 관측 - 85% 이상
  • Threads : 처리량을 넘어가는 색인 / 검색 요청 존재 여부 - Rejected Threads 발생시
  • ** 문제 원인 분석에 사용되는 지표 **
    • Memory Usage : 노드에 설치되어 있는 물리적 메모리 사용량 (JVM Heap과 별개)
    • GC Duration : gc 소요시간
    • GC Rate : Old/Young GC의 발생 주기 파악
    • Disk I/O : 노드에 발생하는 디스킈 연산의 지연 시간
    • Latency : 검색과 색인 요청에 소요되는 시간
    • Rate : 검색과 색인 요청이 인입되는 양

• 문서를 분석하고 저장하는 과정

간단한 색인 과정

• 인덱스 존재 여부 판단 (없으면 생성)
• 매핑 정보 존재 여부 판단 (없으면 동적매핑)
• 매핑 올바른지 판단(type등 문제 있을경우 에러 발생)
• inverted index 생성
• 프라이머리 샤드에 저장
• 레플리카 샤드에 복사

색인 성능과 샤드 갯수

• 색인 성능을 위해서는 클러스터로서의 이점을 살리고 있는지 확인
• 클러스터의 이점을 살리기 위해서는 적절한 노드 갯수, 성능 및 샤드 갯수 중요
• 처음부터 완벽하게 샤드를 배치할 수는 없음
  • 점차 최적화해야됨
• 샤드 갯수를 최적화한 후, 노드에 대한 Scale-up & Scale-out
  

inverted index : 문자열을 분석한 결과를 저장하고 있는 구조체

간단한 검색 과정

• 검색어 분석 (analyzer 적용해서 토큰 생성)
• inverted index 검색 (생성된 토큰을 inverted index에서 검색)
• 검색 결과 표시

Analyzer의 토큰 생성 과정

• 문자열 입력
• character filter : 특수문자 제거 등 필터링 과정
• tokenizer : 공백 등 특정 기준을 바탕으로 tokenize
• token filter : 대소문자 처리 등 추가 필터링

주의 사항

• 색인 : primary shard에서만 가능
• 검색 : primary & replica 둘 다 가능

text와 keyword

• text : 전문 검색(Full-text search)을 위해 토큰 생성
  • ex) "Hello world" 에 대해 "Hello", "World"로 두 개 토큰 생성
• keyword : Exact Matching
  • ex) "Hello world" 에 대해 "Hello world"라는 하나의 토큰 생성
  • 쪼개는 과정이 없어 색인 속도가 빠름

문자열 필드는 동적 매핑되면 keyword, text 타입 두 개 모두 생성

• lucene 기반 오픈소스 검색 엔진 (json 기반 문서 저장, 검색 및 분석)

Elastic Search 특징

• near real time 검색
• 클러스터 구성
  • 한 대 이상의 노드로 클러스터 구성 - 부하 분산, 장애 대응
• 동적 스키마 생성 - 입력 데이터에 대해 스키마 생성해 줌
  • 정적 스키마도 지정 가능
• RestAPI 기반 인터페이스

클러스터와 노드

• 클러스터 : 노드 여러개 모여서 이룬 군집
• 노드 종류
  • 마스터 노드 :
    • 클러스터 상태, 메타데이터 관리
    • 마스터 노드와 마스터 후보 노드로 구성
  • 데이터 노드 : 문서 색인 및 검색 처리
  • 코디네이팅 노드 : 검색 처리
  • 인제스트 노드 : 색인(저장)되는 문서의 데이터 전처리
• 클러스터이므로 어떤 노드에 어떤 요청을 해도 응답은 동일
• 하지만 더 최적화된 작업을 할 수 있게(본인 역할에 충실하게) 구성해야 함

인덱스와 샤드

• 인덱스 : 문서가 저장되는 논리적인 공간

  • 인덱스 설계 (하나의 큰 인덱스 vs 여러개로 쪼갠 인덱스) 중요
    • 하나의 큰 인덱스 : 관리 리소스 적게 발생 but 쿼리와 문서가 복잡해질 가능성
    • 여러개로 쪼갠 인덱스 : 각각의 경우에 대해 최적화된 쿼리 but 관리 어려움 증대
  • 추천 -> 하나의 인덱스로 단순하게 시작하고 경우에 따라 쪼개기

• 샤드 : 인덱스에 색인되는 문서가 저장되는 공간

  • 하나의 인덱스는 반드시 하나 이상의 샤드 갖음

  • 샤드 종류

    • 프라이머리 샤드 : 문서 저장되는 원본 샤드, 색인 및 검색 성능에 모두 영향

    • 레플리카 샤드 : 프라이머리 샤드의 복제 샤드로써 검색에 영향

      프라이머리 샤드에 문제가 생기면 프라이머리 샤드로 승격함

    • 문서들은 모든 샤드에 고르게 저장됨

    • 인덱스 생성 이후 프라이머리 샤드 갯수는 변경 불가 - 신중하게 설정

전반적으로 네트워크 흐름을 생각하기에 좋은 강의였다. 디테일하게 채우긴 쉽지 않지만, 각 Layer 마다 특징적인 부분들 등에 대해서는 잘 학습할 수 있는 좋은 강의였다.

영상에서의 핵심은 결국

Clean code that works

였던 것 같다.

사실 회사에서의 업무 자체도 클린 코드랑 거리가 멀기도 하고, 그 가치를 알아주는 사람이 많이 없어서 여러모로 아쉬움을 느끼고 있기는 하다.

그래도 그 가치를 나 혼자라도 지켜가야지 싶기는 하다.

강의에서 다뤘던 테스트코드 관련 부분도 인상깊다. 테스트코드를 짜면서 개발하는게 분명 초반에는 생산성에서 약간 이슈가 될 수 있는데 이 부분을 어떻게 해결하는가? 에 대한 명쾌한 해답

테스트 코드를 최대한 빨리 짜자!

뭔가 굉장히 심플하고 웃음이 나지만, 이것보다 명쾌한 해답이 없을 것 같다. 테스트 코드를 짜는 능력을 더 키워야겠다.

이 두 내용이 가장 인상깊었다.

예시코드

@DisplayName("상품 타입이 재고 타입인지 체크한다.")
@CsvSource({"HANDMADE,false", "BOTTLE,true", "BAKERY,true"})
@ParameterizedTest
void containsStockType3(ProductType productType, boolean expected){
    //given
    //when
    boolean result = ProductType.containsStockType(productType);
    //then
    assertThat(result).isEqualTo(expected);
}

8. DynamicTest

   • @TestFactory를 사용하고, 아래에서 여러 행동을 결합하여 특정 시나리오대로 테스트 진행

예시코드

@DisplayName("재고 차감 시나리오")
@TestFactory
Collection<DynamicTest> stockDeductionDynamicTest(){

    //given
    Stock stock = Stock.create("001", 1);

    return List.of(
             DynamicTest.dynamicTest("재고를 주어진 개수만큼 차감할 수 있다.", () -> {
                //given
                int quantity = 1;

                //when
                stock.deductQuantity(quantity);

                //then
                assertThat(stock.getQuantity()).isZero();
            }),
            DynamicTest.dynamicTest("재고보다 많은 수의 수량으로 차감 시도하는 경우 예외가 발생한다.", () -> {
                //given
                int quantity = 1;

                //when //then
                assertThatThrownBy(() -> stock.deductQuantity(quantity))
                        .isInstanceOf(IllegalArgumentException.class)
                        .hasMessage("차감할 재고 수량이 없습니다.");
            })
    );
}

9. 테스트 환경 통일

   • ActiveProfile 차이 등으로 인해 테스트시 SpringBoot 서버가 새로 뜸. 이는 자원 낭비
   • 환경을 통합하는 부모 클래스를 만들고, 이를 상속하게 하면서 이러한 테스트 시 서버 재가동 줄일 수 있음
   • Mock의 경우에는 또 다른 환경이므로 이에 대핸 구분 필요 (Mock을 사용하는 class끼리만 통합)
   • Presentation 계층의 경우 @SpringBootTest가 아닌 @WebMvcTest 사용하므로 통합 어려움, 같은 계층끼리만 통합

10. private 메서드의 테스트

    • 할 필요 없음
    • 외부(호출하는 입장)에서는 private method는 알 필요가 없음
    • private method를 사용하는 쪽에서 test를 진행하면 자연럽게 해당 private method의 검증도 되는 구조
    • 계속 이러한 필요가 생각난다면, 오히려 클래스의 책임 분리에 대해 생각할 떄 일지도..

11. production에서는 필요 없고 test에만 필요한 코드라면?

    • 만들어도 되지만 보수적으로 접근
    • 정말 꼭 필요한 경우에만 작성하자

츨처 - https://www.inflearn.com/course/practical-testing-%EC%8B%A4%EC%9A%A9%EC%A0%81%EC%9D%B8-%ED%85%8C%EC%8A%A4%ED%8A%B8-%EA%B0%80%EC%9D%B4%EB%93%9C/dashboard

• Stub 상태 검증

• Mock 행위 검증

Spark RDD

RDD는 Resilient Distributed Dataset의 줄임말로 분산 데이터 모델이다. RDD는 의존성, 파티션, 연산 함수라는 세 가지 특성이 있다.

1. 의존성

어떤 입력을 필요로 하고 현재의 RDD가 어떻게 만들어지는지 스파크에게 가르쳐 주는 것을 의미 결과를 새로 만들어야 할 경우 스파크는 의존성 정보를 참고하고 연산을 다시 반복하여 RDD 다시 만듦 이를 통해 RDD에 유연성을 부여함

2. 파티션

스파크에게 작업을 나눠서 이그제큐터들에게 분산해 파티션별로 병렬 연산을 할 수 있게 함 이를 통해, 작업의 효율성 증진시킴

3. 연산 함수

RDD에 저장되는 데이터를 Iterator[T]형태로 만들어 주는 연산 함수를 가지고 있음

스파크의 구조 확립

위의 스파크의 핵심 특성들에는 연산 함수 및 연산식의 불투명성 등과 같은 문제가 있었음

이를 해결하기 위해, 스파크 2.x부터 여러 개념을 도입함

• 데이터 분석을 통해 찾은 일상적인 패턴들을 사용하여 연산을 표현(필터링, 선택, 집합연산 등)
• DSL에서 일반적인 연산 집합 사용
• 지원하는 정형화 데이터 타입을 사용하여 데이터를 표 형태로 구성할 수 있게 됨

이렇게, 스파크는 구조 확립을 통해 더 나은 성능, 공간 효율성 및 명료한 표현 등 이득을 얻음 ex) 스파크에게 직접 연산을 명령하기 vs 무엇을 할지(작업 자체)를 명령하기

Spark에서의 분산 데이터와 파티션

스파크에서 실제 물리적인 데이터는 HDFS(HaDoop File System)나 클라우드 저장소에 존재하는 파티션이 되어 저장소 전체에 분산된다. 데이터가 파티션이 되어 물리적으로 분산되며, 스파크는 각 파티션을 메모리의 데이터 프레임 객체로 바라본다. 각 스파크 이그제큐터는 데이터 지역성을 고려하여 네트워크에서 가장 가까운 파티션을 읽도록 태스크를 할당한다. 이러한 파티셔닝을 통해, 네트워크 사용을 최소화하고 효과적인 병렬 처리를 가능하게 해준다.

스파크 어플리케이션 작업 단위별 설명

• 태스크 : 스파크 이그제큐터로 보내지는 작업 실행의 가장 기본적인 단위
• 스테이지 : 서로 의존성을 가지는 다수의 태스크 모음
• 잡(job) : 스파크 액션에 대한 응답으로 생성되는 병렬 연산. 스테이지의 모음.

스파크 연산의 종류

• 트랜스포메이션(transformation)

  이미 불변성의 특징을 가진 원본 데이터를 수정하지 않고 하나의 스파크 데이터 프레임을 새로운 데이터 프레임으로 변형하는 연산 (ex: select(), filter() 등) 트랜스포메이션의 결과는 즉시 계산되는게 아니라 계보(lineage)라 불리는 형태로 기록됨(지연평가) 기록된 리니지는 실행 계획의 후반에 확실한 트랜스포메이션들끼리 재배열 및 결합 등을 통해 최적화

• 액션(action)

  모든 기록된 트랜스포메이션의 지연 평가를 발동시킴. 즉 액션 a가 호출되면 그 전까지 기록된 모든 트랜스포메이션이 실제로 실행되며 데이터에 접근함 (ex, show(), take() 등)

지연 평가를 통해 쿼리의 최적화를 가능하게 하며, 리니지와 데이터 불변성을 통해 데이터 내구성 제공

스파크 연산 - 트랜스포메이션의 종류

• 좁은 의존성 하나의 입력 파티션을 연산하여 하나의 결과 파티션을 내놓는 트랜스포메이션

• 넓은 의존성 groupBy()나 orderBy()를 쓸 경우, 다른 파티션으로부터 데이터를 읽어 들이고 디스크에 쓰는 작업 등

아파치 스파크는 데이터 센터나 클라우드에서 대규모 분산 데이터 처리를 하기 위해 설계된 통합형 엔진임

Apache Spark의 핵심 특징 4가지

1. 속도

• 중간 연산을 위한 메모리 저장소를 지원하여 Hadoop Map-Reduce보다 빠름
• 또한, 질의 연산을 DAG(Directed Acyclic Graph)로 구성하고, DAG 스케쥴러 및 쿼리 최적화 모듈은 효율적인 연산 그래프를 만들고, 클러스터의 워커 노드에서 병렬 수행될 수 있게 함
• 물리엔진 텅스텐(Tungsten)은 whole-stage code generation 기법을 사용하여 실행을 위한 간결한 코드를 생성함
• 디스크 I/O의 제한적 사용

2. 사용 편리성

• 스파크는 데이터 프레임이나 데이터세트 같은 고수준 데이터 추상화 계층 아래 유연한 분산 데이터 세트 RDD(Resilent Distributed Dataset)이라는 자료구조 구축. 단순성 실현함
• 연산(Operation)의 종류로 트랜스포메이션(Transformation)과 액션(Action) 두 가지만 가지고 프로그래밍 할 수 있게 해서 단순성 실현

3. 모듈성

• 스파크 연산은 다양한 타입의 워크로드에 적용 가능하며, 지원하는 프로그래밍 언어(Scala, Python, Java, R)로 표현 가능
• 통합 라이브러리와 여러 컴포넌트들을 제공하고 이를 하나의 엔진 안에서 연동할 수 있음

4. 확장성

• 스파크는 저장보다는 빠른 병렬 연산 엔진에 초점이 맞추어져 있음. (이 점이 저장과 연산을 모두 포함하는 Apache Hadoop과의 차이) 이는 스파크가 수많은 데이터 소스에서 데이터를 읽고, 이를 메모리에서 처리 가능하다는 의미

그 외 특징

• 스파크는 JVM 위에서 동작. 즉 어떤 언어로 작성하든 실제 코드는 고도로 경량화된 바이트코드로 변환됨
• 스파크 SQL은 ANSI SQL 2003 표준과 호환.
• 데이터 과학을 위한 MLlib 존재
• 그래프를 조작하고 그래프 병렬 연산을 수행하게 해주는 GraphX 라이브러리 존재

아파치 스파크의 구성

• 하나의 스파크 애플리케이션은 하나의 드라이버 프로그램으로 구성. 이 드라이버는 Spark Driver임. Spark Driver는 Spark Session을 생성하고,이것이 스파크 클러스터의 분산 컴포넌트들에 접근함

스파크 드라이버(Spark Driver) :

• Spark Session 객체 초기화
• Cluser Manager와 통신하며 스파크 이그제큐터들을 위해 필요한 자원 요청
• 모든 스파크 작업을 DAG 연산 형태로 변환 및 스케줄링하여 각 실행 단위를 태스크로 나눔
• 나눈 태스크들을 스파크 이그제큐터에 분배

스파크 세션(Spark Session) :

• 모든 스파크 기능을 한 군데에서 접근할 수 있는 시작점 제공

클러스터 매니저(Cluser Manager) :

• 스파크 애플리케이션이 실행되는 클러스터에서 자원을 관리 및 할당하는 책임 현재 네 종류의 클러스터 매니저(standalne, hadoop YARN, Mesos, K8s) 지원

스파크 이규제큐터(Spark Executor) :

• 클러스터의 각 워커 노드에서 동작. 드라이버의 프로그램과 통신하며 워커에서 태스크를 실행하는 역할 (대부분 배포 모드에서 노드 당 하나의 이그제큐터만 실행)

일반적인 어플리케이션에서 데이터베이스의 데이터는 비즈니스 상황을 반영함. CRUD가 지속적으로 일어나고, 데이터의 변경이 일어날 것이며 이렇게 데이터가 변경될 경우 이전 데이터는 백업이 있고 하겠지만 비즈니스 상 크게 중요하지 않음(현재의 데이터가 중요하니까)

하지만, 특정 몇 몇 경우에는 이전의 상황이 분명히 필요함 (ex: 은행 잔고의 경우 x일 전에 y원만큼 변경되어 z원의 잔고 존재라는 정보가 중요할것임)

이 때, 필요한것이 Event Sourcing Pattern

이벤트소싱 패턴은 현재 상태는 저장하지 않고 이벤트만 저장함 각 이벤트는 시스템 내 특정 엔티티에 대한 변경 사항과 사실 정보로 구성됨

이벤트의 가장 큰 특징은 불변성(immutable) 데이터에 대한 VCS라고 생각하면 유사

이벤트 소싱 패턴의 장점

• 시각화
• 감사
• 검증
• 높은 쓰기 성능

이벤트 소싱 패턴에서 이벤트를 저장하고 표현하는 방식

• 데이터베이스에서 각 이벤트를 개별 레코드로 저장
• 메세지 브로커를 통한 이벤트 저장

• 대부분의 DB와 달리 다량의 이벤트를 처리하는데 최적화

• 이벤트 스트림에서 복잡한 쿼리를 처리하기는 힘듦

맨 위의 은행 잔고 예시에서, 이벤트 소싱 패턴을 적용했다면? 잔고를 저장하는게 아니라 거래 내역을 저장하여 이를 통해 계산하는 방식으로 데이터를 얻어냄

이벤트 소싱 패턴에서 모든 이벤트를 저장해놨다가, 특정 시점에서의 값을 구하기위해 처음부터 이벤트를 쭉 읽어내려오며 계산하는것은 사실 좀 비효율적임

이를 해결하기 위한 방법들이 존재

• 특정 시점의 snapshot 만들기

• 스냅샷 이후부터 계산하면 됨

• CQRS 패턴 적용

• 명령과 쿼리를 분리하여 쿼리 서비스에게 조회만 시키면서 조회 성능을 향상시킴. (업계에서 많이 쓰이는 방식, 물론 최종적 일관성만 보장된다는 단점이 있기 때문에 주의해야 함)

마이그레이션 상황 오래된 모놀리식 어플리케이션을 마이크로서비스로 마이그레이션하려 함

보통 이런 상황에서는

• 시스템 및 아키텍쳐 최신화
• 어플리케이션 현대화

이 둘 모두를 한꺼번에 하려고 할 것임

하지만 마이그레이션을 진행한다고 해서 모든 개발을 중단할 순 없음. 즉 운영중인 서비스는 계속 고도화 및 확장 등이 일어날것임

이렇게 될 경우, 하나씩 마이크로서비스를 만들어가면서 현재 돌아가는 모놀리식 어플리케이션과 연동시킬거임 (마이크로서비스가 전부 모놀리식 어플리케이션을 대체할 때 까지)

하지만, 한번에 완전히 전환하지 않으면 마이크로서비스들이 저 현대화 전 레거시 어플리케이션에 의존하게됨(API, Data 등에 대해서) 이러면 어플리케이션 현대화의 의미가 퇴색됨

이러한 상황을 방지하기 위해 손상 방지 계층 패턴이 존재함

이 패턴은 이전 시스템과 새 시스템의 어댑터 기능을 하는 서비스를 통해 새로 작성한 마이크로서비스와 모놀리식(레거시) 어플리케이션이 직접 연결되는것을 방지함

서로 통신을 할 때 무조건 손상 방지 계층을 지나게 함으로써, 이 둘 간 의존성을 없에게 되는게 핵심

일반적으론 마이그레이션 완료 후에 어댑터를 없엘거임 하지만 마이그레이션 이후에도 레거시 코드는 남아있을 수 있음. (완전히 레거시 코드를 제거할 수 없는 상황도 존재할 수 있음)

이런 상황에서, 손상 방지 계층을 남긴다면, 이를 통해 레거시 시스템과 새로운 시스템이 통신하게 하여 새로운 시스템의 손상을 방지함

만능은 아님. 생각해 볼 요소들이 있음

1. 손상 방지 어댑터 또한 개발, 테스트 ,배포되어야 하는 서비스임 개발해야 할 요소가 증가함

2. 손상 방지 계층의 존재 자체가 통신에 하나의 계층이 추가되는것을 의미 이는 즉 추가적 오버헤드의 발생을 의미함

모든 서비스에는 핵심 기능이 있음(서비스의 존재 이유) ex) 고객 서비스 : 고객 정보 저장 ex) 결제 서비스 : 결제 내역 저장 및 결제 처리 등

하지만, 각 서비는 핵심 기능 외에 다른 기능도 해야함 예를 들면 metric을 모니터링 서비스에 전송한다던지.. 구성 파일을 가져와 파싱하고 비지니스 로직에 적용한다던지..

동일한 라이브러리를 다른 언어로 구현하는것은 상호 호환성과 일관성을 떨어뜨림 (데이터 유형 등 차이로 다양한 문제 발생 가능)

이러한 문제를 해결하는것이 Sidecar pattern

애플리케이션에 필요한 추가 기능을 메인 애플리케잇녀과 동일한 서버에 개별 프로세스 (혹은 컨테이너)로 실행 사이드카 프로세스와 메인 프로세스를 분리하는것(같은 서버에서)

개별 프로세스로 사이드카 프로세스를 만들면, 하나의 언어로 사이드카를 만들고 이를 재활용할 수 있음

핵심 서비스의 구현 및 테스트에 집중할 수 있음 (공통 기능 및 기타 기능을 사이드카에 몰아놨으니까)

Ambassador(특사) pattern

ambassador pattern은 sidecar pattern의 special case임

ex)서비스를 대신해 모든 네트워크 활동을 전송하는 사이드카 이 경우 서비스 책임 범위 밖으로 복잡한 네트워크 통신 로직을 빼낼 수 있음

-> 이렇게하면 진짜 핵심 비즈니스 로직만 애플리케이션 프로세스에 남게 됨

모든 서비스의 통신이 같은 위치에 있는 특사 사이드카에서 이루어지므로 쉽게 네트워크 통신 실행 및 분산 추적 가능

결국 핵심은 공통 기능과 핵심 기능의 분리를 코드 레벨이 아닌 아키텍쳐 레벨에서 구현한 것이라 볼 수 있을 것 같다. (약간 Spring AOP 생각난다.)

Command and Query Responsibility Segregation

데이터를 데이타베이스에 저장하는 일반적 시스템은 데이터에 대한 작업을 두 유형으로 나눔

• 명령 -> DB의 데이터를 변경시킴(추가, 수정 및 삭제) (DB에 레코드 삽입, 업데이트 및 삭제 등 변경 발생)

• 쿼리 -> DB의 데이터를 조회함 (DB 데이터의 변경 없음)

CQRS는 시스템의 명령 파트와 쿼리 파트를 명확히 분리함 서비스를 (1)명령 서비스, (2) 쿼리 서비스 라는 두 가지 서비스로 분리하고, 각자 DB를 가지게 함. 이를 통해, 쿼리 서비스(조회)를 고성능으로 유지할 수 있음.

새로운 비즈니스 로직이 추가되어도 명령 서비스에만 변경이 일어나고, 쿼리 서비스는 변경되지 않으므로 테스트하거나 배포할 필요가 없음

CQRS Pattern을 적용할 시 장점:

• 시스템을 명령, 쿼리 둘 모두에 대해 최적화 할 수 있다.

• 그러므로, 읽기와 쓰기 모두 빈번한 시스템에서 특히 중요하다

• 높은 확장성을 가지게 된다.
  

CQRS Pattern 적용 시 발생할 수 있는 문제점

• 최종적 일관성만 보장됨. 엄격한 일관성이 필요할 경우 적합하지 않음

• 작업 과정 중간에 명령 서비스 DB와 쿼리 서비스 DB가 다른 상황이 존재할 수 있음

• 시스템에 오버헤드와 복잡성을 추가시킴

• CQRS를 통해 얻는 성능상의 이점이 이러한 부분보다 큰지 검증 후 사용

중요한 부분은, 명령 서비스에서 데이터가 수정되었을 때, 수정된 데이터를 어떻게 쿼리 서비스에 전파하여 동기화시킬 것인지 이를 위해서, 명령 파트에 데이터 변형 요청이 발생할 때 마다 쿼리 서비스가 이를 수신하게 함

동기화 방법

• 명령 서비스와 쿼리 서비스의 사이에 메세지 브로커 두기

• 트랜잭셔널 아웃박스 패턴을 적용하여 단일 트랜잭션에서 DB와 메세지 브로커 모두에 메세지 전달

• 서비스형 함수(function as a service) 사용하기

• 서비스형 함수를 통해 명령 서비스 DB의 변경을 감지

데이터 집약적인 어플리케이션이 가지는 문제들이 있음

1. 성능문제 여러 테이블이나 여러 데이터베이스간 걸쳐있는 복잡한 쿼리를 실행할 경우 시간도 오래 걸리게 됨

2. 효율성과 비용 대용량 데이터 셋을 읽거나 종합 및 변환하는 복잡한 쿼리를 반복적으로 실행하면 쿼리 실행에 들어가는 리소스 낭비가 심함

이를 해결하기 위해 구체화된 뷰(Materialized View) pattern 사용

구체화된 뷰 패턴은 읽기 전용 테이블을 생성하여, 최적화하려는 쿼리의 결과값을 미리 채워넣음

고려사항

• 구체화된 뷰 테이블이 필요로하는 추가 공간을 고려해야 함

  -> 공간과 시간의 trade-off. 평범하지만 cloud 환경에서는 비용 문제를 생각해야함

• 구체화된 뷰를 어디에 저장하는가?

  -> 원본 데이터가 있는 DB의 개별 테이블에 저장하는게 일반적 -> In-memory Cache 에 저장할 수도 있음(Redis 등)

Message broker에게 메세지를 주는 대신, DB에 Outbox table을 두고 여기에 새 record로써 메세지를 추가하게 됨 (중요 사항: 이 작업은 단일 데이터베이스 트랜잭션에서 수행됨 -> atomicity가 보장됨)

이 후, Message Relay라는 서비스를 둠. 이 서비스는 Outbox table에 record가 추가됐을 경우, 이를 가져와서 Message broker에게 전송한 후, table에서 해당 record를 삭제함

이 패턴의 문제점

• 이벤트 중복의 가능성: Message Relay(Sender)가 Outbox table에서 추가된 record 형태의 메세지를 읽고 이를 Message broker에 보낸 상태에서, record를 삭제하기 전에 재시작된다면 다시 record를 Message broker에 보낼 수 있고, 이는 이벤트의 중복이 됨 (At Least Once Delivery Semantics)

  -> Message Broker 이후 Scheduling Service를 통해 멱등 조작을 실현하면 됨 -> record 마다 고유한 id를 통해 중복 여부를 검사함

-> 시퀀스 id를 record에 부여하고, 이 순서대로 해결