[AWS] Athena: DynamoDB 통합

DynamoDB는 날강도들 가득한 AWS에서 가장 강력하고 비용 효율적인 DB 서비스다.
RDB만큼 유연한 검색기능을 제공하진 않지만, 무제한 쓰기 확장이 가능하고, 비용도 성능 대비 저렴한 편이기 때문이다. 그래서 로그 같은 APPEND Only 데이터 수집 창고로 쓰기 좋다.

그리고 DynamoDB의 또 하나의 장점은, Athena & S3 파이프라인과도 아주 잘 통합된다는 것이다.
서비스가 DynamoDB에 데이터를 쌓으면, 그걸 S3에 parquet으로 말아서 올리고, Athena로 대량 집계를 날리는 일련의 패턴을 편리하게 구성할 수 있다.

아키텍쳐 선택지

Athena에서 DynamoDB를 소스로 사용하는 방법은 여러가지가 있다.

1. 직구로 꽂기

가장 단순한 사용 방식은 다음과 같다.
DynamoDB Connector를 사용해서 직통으로 연결하는 것이다.

이건 성능이나 비용상의 이점이 거의 없다.
그냥 DynamoDB에 대신 READ 날려서 SQL Query에 맞춰서 가공을 하는 것에 불과하고, DynamoDB의 성능 한계를 그대로 받는다. 비용도 DynamoDB RCU 기반으로 뜯는다.
이건 작은 데이터에 대해서 JOIN 하는 용도로만 쓰거나 해야한다.

2. S3 Export Only

그 다음으로 나은 방법은 다음과 같다.
DynamoDB는 PITR 백업 기반으로 S3에 export하는 기능을 제공하는데, 그러고 나면 아테나에서 바로 꽂아서 읽을 수 있다.

근데 이것도 사실 최선이라고 할 수는 없다.
아테나 조회 패턴에 최적화되어있지도 않고, 공간도 불필요하게 크게 먹기 때문이다.
성능과 비용효율성을 원한다면 parquet으로 말아놓는 구성이 필요하지만, 이건 그냥 flat하게 펼쳐버린다.

3. S3 Export & CTAS (parquet)

Athena CTAS라는 기능을 쓰면 Athena만을 위한 최적화된 구조로 재생성시킬 수 있다.

4. Best Practice?

Best Practice로 간주되는 방법은 다음과 같다.

증분형 S3 export를 통해서 S3에 붓고, 그걸 Athena를 사용해서 재생성하는 것이다.
ETL 제어에는 Step Function이 가장 단순하고 편리한 편이다.

비용

https://aws.amazon.com/ko/dynamodb/pricing/on-demand/

비용 발생 요소는 몇가지가 있다.
PITR이라는 복구 데이터를 유지해야 그걸로 S3에 보내는건데, 그것도 백업본 전체 용량 기준으로 요금을 뜯는다.

전체 백업이 200GB면 40달러 정도 뜯는 셈이다.
근데 최소한도인 1일로 잡으면 그 날의 변경분만 보존될테니, 그렇게 크게 들진 않을 것이다.

내보내는 행위 자체에도 크기당 비용이 발생한다.

100GB 정도 된다면 만원쯤 나오는 셈이다.

그리고 당연히 S3 비용도 발생한다.
저장 비용은 당연히 따로고, parquet으로 재생성하는 과정에서 S3 표준 읽기/쓰기 비용이 발생한다.

사용해보기

DynamoDB 테이블을 직접 만들고, 과정을 거치면서 사용법을 익혀보자.

적당히 테이블 하나 만들고

백업 옵션인 PITR을 켜야 한다.
이건 본질적으로 백업 데이터라서, 따로 백업정책이 필요없고 S3 Export용으로만 쓸거라면 1일로 해도 된다.

그리고 적당히 데이터를 넣는다.

S3 Export

이제 S3로 부을 준비는 됐다.
Dynamo에 보면 내보내기 탭이 있는데,

여기서 내보낼 S3 버킷 고르고 트리거하면 된다.
전체 내보내기와 증분(부분) 내보내기가 있는데, 일단은 전체로 먼저 해본다.

그리고 이 내보내기 자체는 비동기 API라서 즉시 되진 않는다.

좀 기다리면 성공으로 끝날텐데

그럼 목적지에 이런 파일들이 지저분하게 막 떨어져있을 것이다.
이게 필요한 데이터들이다.

매핑용 External 테이블 구성하기

자. 이제 준비는 거의 됐다.
Athena로 이동해서 매핑 테이블을 구성해준다.
방금 Export된 json.gz들이 있는 경로를 넣고, 어떤 필드들이 있는지 체크해서 키 목록을 세팅해주면 된다.

CREATE EXTERNAL TABLE 외부테이블구성 (
    Item struct<
        키:   struct<S:타입>,
        ...
    >
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION 's3://버킷/경로/dynamo/AWSDynamoDB/?/data/';

이것 자체는 기존 데이터에 메타데이터만 추가해주는거라서, 비용이나 성능을 먹지는 않는다.

그럼 일단 조회를 해볼 수는 있을 것이다.

우린 여기서 한발 더 나아가야 한다.
athena의 CTAS 기능을 사용해서 parquet로 재생성해준다.

CREATE TABLE access_logs_parquet
WITH (
  format = 'PARQUET',
  parquet_compression = 'SNAPPY',
  external_location = 's3://저장경로/'
) AS
SELECT
  Item.id.S   AS id,
  Item.ip.S   AS ip,
  Item.path.S AS path
FROM 임시테이블;

그럼 실제로 최종 목적지에 압축된 parquet 엔트리가 생성될 것이고

쿼리도 날려볼 수 있을 것이다.

잘 들어간 것을 볼 수 있다.
쿼리 날리는건 그냥 이렇게 하면 된다.

DynamoDB의 증분 Export (Incremental Export)

근데 방금 사용한 방식은 잘 동작하긴 하지만, 확장성이 좋은 방식은 아니다.
매번 전체 데이터를 가져와서 밀어넣는 것은 너무 오래걸리고, 비용도 비용대로 먹는다.

다행히 DynamoDB는 이런 딜레마를 잘 해소해준다.
DynamoDB의 백업 시스템 자체가 타임스탬프 기반으로 필터링이 되는 구조로 되어있어서, "몇시간 전부터의 데이터"나 "어제부터의 데이터" 같은 식으로 부분적 EXPORT가 잘 되기 때문이다.

이번에는 행을 하나 추가한 다음에 동기화를 구성해보자.

id=4444를 추가하고

증분 내보내기를 선택해서 트리거한다.

그럼 마찬가지로 json.gz가 잔뜩 떨어진다.

근데 여기서부턴 좀 다르다.
전체 내보내기는 진짜 전체 데이터만 가져오는 거라서 구조가 단순한데, 증분 내보내기는 UPDATE/DELETE 같은 상태 정보도 함께 포함되기 때문이다. 그 시기의 "모든 변경"을 가져온다.

CREATE EXTERNAL TABLE access_log_incremental (
    Metadata struct<
        WriteTimestampMicros: string
    >,
    Keys struct<
        id: struct<S:string>
    >,
    NewImage struct<
        id:   struct<S:string>,
        ip:   struct<S:string>,
        path: struct<S:string>
    >,
    OldImage struct<
        id:   struct<S:string>,
        ip:   struct<S:string>,
        path: struct<S:string>
    >
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION 's3://버킷/경로/dynamo/AWSDynamoDB/?/data/';

그래서 이런 식으로 뭐가 좀 많다.
INSERT/UPDATE는 NewImage를 읽어서 넣어야 하고, DELETE는 Keys만 전달된다.

여기서부터는 기능 구조에 따라서 제약조건이 좀 달라진다.
APPEND-ONLY의 단순한 시스템이라면, 그냥 NewImage만 걸러서 기존 CTAS 테이블에 insert하면 된다.

그럼 자동으로 parquet 덩어리가 추가될 것이고

조회도 잘 될 것이다.

근데 UPDATE/DELETE를 고려해야 한다면 좀 복잡해진다.

사실 지금 위의 쿼리도, 기존 parquet에 들어있던 ID와 동일한 INSERT가 들어온다면 중복 ID가 발생할 수 있다는 단점이 있다. 원본 데이터부터가 NEW INSERT ONLY라는 전제 하에서만 잘 동작하는 구조다.

IceBerg를 사용한 UPDATE/DELETE

여기서 문제가 되는 것은 parquet 자체가 수정을 위한 데이터 규격이 아니기 때문이다.
그저 데이터를 압축해서 효율적으로 관리하기 위한 포맷에 불과하다.

그래서 변경되는 데이터셋에 대해 수정과 삭제를 고려해서 parquet을 관리해야 한다면, iceberg 엔진을 사용하는 편이 권장된다.
다음과 같은 식으로 테이블을 만들면 iceberg의 파티션 관리 능력을 가져갈 수 있다.

CREATE TABLE access_log_iceberg
WITH (
  table_type = 'ICEBERG',
  format = 'PARQUET',
  location = 's3://.../iceberg/access_log/',
  is_external = false
) AS
SELECT id, ip, path FROM access_log_parquet;

그러면 이제 MERGE 구문을 통해서 파티션을 재구성하는 것이 가능해진다.
DELETE나 UPDATE를 반영할 수 있다.

MERGE INTO access_log_iceberg t
USING (
  SELECT
    COALESCE(NewImage.id.S, Keys.id.S) AS id,
    NewImage.ip.S   AS ip,
    NewImage.path.S AS path,
    NewImage IS NULL AS is_deleted,
    CAST(Metadata.WriteTimestampMicros AS bigint) AS change_ts,
    ROW_NUMBER() OVER (
      PARTITION BY COALESCE(NewImage.id.S, Keys.id.S)
      ORDER BY CAST(Metadata.WriteTimestampMicros AS bigint) DESC
    ) AS rn
  FROM access_log_incremental
) s
ON t.id = s.id AND s.rn = 1
WHEN MATCHED AND s.is_deleted THEN DELETE
WHEN MATCHED THEN UPDATE SET ip = s.ip, path = s.path
WHEN NOT MATCHED AND NOT s.is_deleted THEN INSERT (id, ip, path) VALUES (s.id, s.ip, s.path);

파티션 전략

Athena를 비용효율적으로, 그리고 더 빠르게 사용하기 위해서는 파티션 전략을 잘 잡아야 한다.
이건 별도 포스트로 갈음한다.
https://blog.naver.com/sssang97/223716323937

참조
https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/S3DataExport.HowItWorks.html

https://docs.aws.amazon.com/athena/latest/ug/connectors-dynamodb.html