Dynamodb enhanced client

dynamodb enhanced client

기존에 dynamodb를 조작하기 위해서 aws에서 제공되던 sdk를 사용할 때 DynamodbDBMapper를 사용해서 객체-테이블 매핑하여 질의해서 사용하였는데 생각보다 사용하기 번거로웠던 기억이 있었다. (마지막으로 프로젝트에서 사용한지 벌써 2년이 지나서 자세한 불편내용은 기억이 나진 않지만 사용에 편리하지 않았던 기억은 명확하게 남아있다.)

 

그래서 이번에 dynamodb를 프로젝트에서 사용하게 되었을 때 조금 더 편하게 사용할 수 있는 방법이 없을까 찾아봤고 dynamodb sdk for Java version 2에서 제공되는 DynamoDb Enhanced를 사용해보기로 했다.

 

 

사용법

gradle import

plugins {
    id 'org.springframework.boot' version '2.7.2'
    id 'io.spring.dependency-management' version '1.0.12.RELEASE'
    id 'java'
}

group = 'com.wedul'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = '17'

configurations {
    compileOnly {
        extendsFrom annotationProcessor
    }
}

repositories {
    mavenCentral()
    maven { url 'https://s3-us-west-2.amazonaws.com/dynamodb-local/release' } // for DynamoDBLocal Lib -- 1
}

dependencies {
    implementation 'software.amazon.awssdk:dynamodb-enhanced:2.17.235'  -- 2
    implementation 'com.amazonaws:DynamoDBLocal:1.11.119'               -- 3
    testImplementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'io.github.ganadist.sqlite4java:libsqlite4java-osx-aarch64:1.0.392' -- 4

    implementation 'org.springframework.boot:spring-boot-starter-web'
    compileOnly 'org.projectlombok:lombok'
    annotationProcessor 'org.projectlombok:lombok'
    testCompileOnly 'org.projectlombok:lombok'
    testAnnotationProcessor 'org.projectlombok:lombok'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
}

tasks.named('test') {
    useJUnitPlatform()
}

1, 3. dynamodb를 로컬에서 테스트하기 위해서 사용하는 dynamodb-local 라이브러리를 사용하기 위한 설정으로 아래에서 사용에 대해 자세히 설명한다.

2. dynamodb-enhanced client를 사용하기 위해 의존성을 추가한다.

4. 1,3번과 동일하게 dynamodb-local에서 사용되는 sqlite4때문에 추가된 내용인데 아래에서 자세히 다루자.

 

 

테이블

@Getter
@Setter
@DynamoDbBean -- 1
@NoArgsConstructor
public class Product {

    private Long productId;
    private Long sellerId;
    private Long rating;
    private Long ttl;

    @Builder
    public Product(Long productId, Long sellerId, Long rating, Long ttl) {
        this.productId = productId;
        this.sellerId = sellerId;
        this.rating = rating;
        this.ttl = ttl;
    }

    @DynamoDbPartitionKey -- 2
    public Long getProductId() {
        return productId;
    }

    @DynamoDbSecondaryPartitionKey(indexNames = "seller_id") -- 3
    public Long getSellerId() {
        return sellerId;
    }

    @DynamoDbSortKey -- 4
    public Long getRating() {
        return rating;
    }
}

1. dynamodb enhanced client에서는 TableSchema를 사용해서 클래스를 테이블에 매핑하도록 제공한다. 만약 선언하지 않고 접근해서 사용하려고 하면 에러가 발생한다.

A DynamoDb bean class must be annotated with @DynamoDbBean

2. dynamodb에서 사용되는 partition key 필드에 애노테이션을 붙인다.

3. dynamodb에서는 partition key, sort key를 사용할 수 있지만 추가적인 조건으로 별도 키를 추가하기 위해서 GSI를 제공하는데 추가할 새로운 index 필드에 해당 애노테이션을 붙이는 것으로 해당 필드가 파티션키로 사용된 인덱스 이름을 함께 기재해준다.

4. dynamodb에서 사용하는 sort key를 추가한다.

 

 

ProductQuery

- product 테이블에 질의를 위한 필드와 쿼리를 정의해놓은 클래스

import lombok.Builder;
import lombok.Getter;
import software.amazon.awssdk.enhanced.dynamodb.Expression;
import software.amazon.awssdk.services.dynamodb.model.AttributeValue;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

@Getter
public class ProductQuery {

    private final int limit;
    private final Long sellerId;
    private final Long lastReadProductId;
    private final Long rating;

    @Builder
    public ProductQuery(int limit, Long sellerId, Long lastReadProductId, Long rating) {
        this.limit = limit;
        this.sellerId = sellerId;
        this.lastReadProductId = lastReadProductId;
        this.rating = rating;
    }

    public Map<String, AttributeValue> exclusiveStartKey() { -- 1
        if (Objects.nonNull(lastReadProductId)) {
            Map<String, AttributeValue> exclusiveStartKeyMap = new HashMap<>();
            exclusiveStartKeyMap.put("sellerId", AttributeValue.builder().n(String.valueOf(sellerId)).build());
            exclusiveStartKeyMap.put("productId", AttributeValue.builder().n(String.valueOf(lastReadProductId)).build());
            exclusiveStartKeyMap.put("rating", AttributeValue.builder().n(String.valueOf(rating)).build());
            return exclusiveStartKeyMap;
        }
        return null;
    }

    public Expression getFilterExpression() { -- 2
        Map<String, AttributeValue> expressionValue = new HashMap<>();

        if (Objects.nonNull(sellerId)) {
            expressionValue.put(":sellerId", AttributeValue.builder()
                            .n(String.valueOf(sellerId))
                    .build());
            return Expression.builder()
                    .expression(":sellerId = sellerId")
                    .expressionValues(expressionValue)
                    .build();

        }
        return null;
    }
}

1. dynamodb는 rdb처럼 페이지방식으로 조회가 불가능하고 마지막으로 읽었던 키를 전달하여 다음 데이터를 얻어오는 식으로 사용된다. 그렇게 하기 위해서는 테이블에 생성한 키값을 모두 사용해야한다. 그래서 만약 sortKey로 사용한 rating값이 주기적으로 바뀌게 된다면 서비스에서 다음페이지를 읽어오지 못하는 문제가 발생할 수 있다. 그래서 리스팅에서 dynamodb를 사용하는건 어느정도 제약이 따른다. (scan방식을 사용하면 되지 않냐고 말할 수 있으나 scan방식은 정렬이 안된다 ㅜ)

2. 특정 attribute기반으로 조회하고자할 때 사용하려고 만든 메소드로 여기서는 sellerId값이 있을 경우 sellerId로 필터해서 데이터를 가져오도록한다.

 

 

processor

- 전달받은 ProductQuery를 통해서 실제 ddb에 질의하고 데이터를 받아오고 저장하는 기능을 하는 processor이다.

@Component
public class ProductProcessor {

    public static final String TABLE_NAME = "product";
    private final DynamoDbTable<Product> dynamoDbTable;
    private final DynamoDbIndex<Product> sellerIdIndex;

    public ProductProcessor(DynamoDbEnhancedClient dynamoDbEnhancedClient) {
        this.dynamoDbTable = dynamoDbEnhancedClient.table(TABLE_NAME, TableSchema.fromBean(Product.class)); -- 1
        this.sellerIdIndex = dynamoDbTable.index("seller_id"); -- 2
    }

    public List<Product> findQuery(ProductQuery query) { -- 3
        QueryEnhancedRequest queryEnhancedRequest = QueryEnhancedRequest.builder()
                .queryConditional(QueryConditional.keyEqualTo(Key.builder()
                        .partitionValue(query.getSellerId())
                        .build()
                ))
                .exclusiveStartKey(query.exclusiveStartKey())
                .limit(query.getLimit())
                .scanIndexForward(false)
                .build();
        return sellerIdIndex.query(queryEnhancedRequest).stream().flatMap(d -> d.items().stream()).collect(Collectors.toList());
    }

    public void save(Product product) { -- 4
        dynamoDbTable.putItem(product);
    }

}

1. dynamodb enhanced client를 사용해서 클래스와 매핑된 객체를 내보내고 이걸 통해서 질의를 할 수 있다.

2.dynamodb enhanced client를 사용해서 secondary index에 매핑된 객체를 받아서 해당 인덱스로 질의할 수 있는 객체를 반환한다.

3. 전달받은 query로 데이터를 찾는 로직을 담긴 메스드로 queryCondition에서는 키를 사용한 데이터를 조회하고 exclusiveStartKey에서는 마지막으로 읽은 키의 데이터를 전달하여 다음에 읽을 데이터를 전달받기위해 사용한다. limit는 우리가 rdb에서 사용한 것과 다르게 결과는 limit보다 많이 리턴된다. 하지만 실제 쿼리 질의에 사용된 결과값을 제한할 수 있다. scanIndexForward는 sort key기준으로 true면 asc, false면 desc로 제공한다.

4. 데이터를 저장하는 기능을 제공한다.

 

 

Client Config

package com.example.dynamodbenhanced.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.context.annotation.Profile;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
import software.amazon.awssdk.core.retry.RetryPolicy;
import software.amazon.awssdk.enhanced.dynamodb.DynamoDbEnhancedClient;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;

import java.net.URI;
import java.net.URISyntaxException;

@Configuration
public class DynamoDbClientConfig {

    @Bean
    @Primary
    @Profile({"local-beta", "beta", "stage", "prod"})
    public DynamoDbClient dynamoDbClient() {
        return DynamoDbClient.builder()
                .overrideConfiguration(createClientConfiguration())
                .region(Region.AP_NORTHEAST_2)
                .build();
    }

    private static ClientOverrideConfiguration createClientConfiguration() { -- 1
        return ClientOverrideConfiguration.builder()
                .retryPolicy(RetryPolicy.none())
                .build();
    }

    @Bean
    public DynamoDbEnhancedClient dynamoDbEnhancedClient(DynamoDbClient dynamoDbClient) { -- 3
        return DynamoDbEnhancedClient.builder()
                .dynamoDbClient(dynamoDbClient)
                .build();
    }

    @Bean
    @Primary
    @Profile({"test", "local"})
    public DynamoDbClient testDynamoDbClient(EmbeddedDynamoDbConfig embededDynamoConfig) throws URISyntaxException { -- 2
        return DynamoDbClient.builder()
                .overrideConfiguration(createClientConfiguration())
                .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create("local", "local")))
                .endpointOverride(new URI("http://localhost:" + embededDynamoConfig.getPort()))
                .region(Region.AP_NORTHEAST_2)
                .build();
    }
}

1. client에 사용될 여러 설정을 재정의한다. 그중 retryPolicy만 재시도 하지 않도록 NONE으로 설정했다.

2. DynamoDbClient를 테스트 용도로 생성했다. dynamodb local port인 8000으로 endpoint를 재정의했다.

3. DynamoDbClient를 사용해서 DynamoDbEnhancedClient를 생성했다.

 

 

테스트

- dynamodb local을 사용해서 테스트를 진행하는데 jojoldu님 블로그에서 볼수 있듯이 sqlite문제가 있어서 동일하게 코드를 추가했다. 하지만 여기서 한가지 더 문제가 있었는데 현재 내가 사용하고 있는 노트북은 m1 노트북이었기에 load하려는 sqlite4 libary가 맞지 않아서 아래와 같은 에러가 발생했다.

cannot load library: java.lang.UnsatisfiedLinkError: Can't load library

해결 방법은 stackover-flow에서 찾았는데 사용할 라이브러리를 추가해주는 방식으로 해결했다. (이 방법으로도 안되면 이방식 참고 https://github.com/aws-samples/aws-dynamodb-examples/issues/22#issuecomment-501492802 )

implementation 'io.github.ganadist.sqlite4java:libsqlite4java-osx-aarch64:1.0.392'

package com.example.dynamodbenhanced.domain;

import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.ActiveProfiles;
import software.amazon.awssdk.enhanced.dynamodb.DynamoDbEnhancedClient;
import software.amazon.awssdk.enhanced.dynamodb.DynamoDbTable;
import software.amazon.awssdk.enhanced.dynamodb.TableSchema;
import software.amazon.awssdk.enhanced.dynamodb.model.CreateTableEnhancedRequest;
import software.amazon.awssdk.enhanced.dynamodb.model.EnhancedGlobalSecondaryIndex;
import software.amazon.awssdk.services.dynamodb.model.Projection;
import software.amazon.awssdk.services.dynamodb.model.ProjectionType;
import software.amazon.awssdk.services.dynamodb.model.ProvisionedThroughput;

import java.util.Arrays;
import java.util.List;

import static com.example.dynamodbenhanced.domain.ProductProcessor.TABLE_NAME;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.*;

@SpringBootTest
@ActiveProfiles({"test"})
class ProductProcessorTest {

    @Autowired
    private ProductProcessor sut;

    private static DynamoDbTable<Product> productDynamoDbTable;

    @BeforeEach
    void setupAll(@Autowired DynamoDbEnhancedClient dynamoDbEnhancedClient) {
        productDynamoDbTable = dynamoDbEnhancedClient.table(TABLE_NAME, TableSchema.fromBean(Product.class));
        productDynamoDbTable.createTable(CreateTableEnhancedRequest.builder()
                        .provisionedThroughput(ProvisionedThroughput.builder() -- 1
                                .readCapacityUnits(10L)
                                .writeCapacityUnits(10L)
                                .build())
                        .globalSecondaryIndices(Arrays.asList(EnhancedGlobalSecondaryIndex.builder() -- 2
                                        .indexName("seller_id")
                                        .projection(Projection.builder().projectionType(ProjectionType.ALL).build())
                                        .provisionedThroughput(ProvisionedThroughput.builder()
                                                .readCapacityUnits(10L)
                                                .writeCapacityUnits(10L)
                                                .build())
                                .build()))
                .build());
    }

    @AfterEach
    void afterAll() {
        productDynamoDbTable.deleteTable();
    }

    @Test
    @DisplayName("sellerId로 원하는 상품을 조회한다.")
    void returnProduct_bySellerId() {
        // given
        Long targetSellerId = 1L;
        Product foundProduct1 = Product.builder()
                .productId(1L)
                .sellerId(targetSellerId)
                .rating(5L)
                .ttl(3L)
                .build();

        Product foundProduct2 = Product.builder()
                .productId(2L)
                .sellerId(targetSellerId)
                .rating(3L)
                .ttl(4L)
                .build();

        Product notFoundProduct1 = Product.builder()
                .productId(3L)
                .sellerId(33L)
                .rating(5L)
                .ttl(6L)
                .build();

        sut.save(foundProduct1);
        sut.save(foundProduct2);
        sut.save(notFoundProduct1);

        // when
        List<Product> result = sut.findQuery(ProductQuery.builder()
                        .sellerId(targetSellerId)
                        .limit(2)
                .build());
        // then
        assertThat(result.size()).isEqualTo(2);
        assertThat(result.get(0)).usingRecursiveComparison().isEqualTo(foundProduct1);
        assertThat(result.get(1)).usingRecursiveComparison().isEqualTo(foundProduct2);
    }

}

1. 테스트에 사용할 테이블 throuput을 정의해준다.

2. 테이블 생성하면서 같이 추가될 secondary index (seller_id)를 추가해주고 테이블과 동일하게 throuput도 같이 설정해준다.

 

결과는 성공

 

 

조금 더 조작하기가 편해졌다. 하지만 ddb 설계상 일부 상황에서는 사용하기가 어려워서 조금 아쉽다.

자세한 사용법은 공식 github 참고 : https://github.com/aws/aws-sdk-java-v2/blob/master/services-custom/dynamodb-enhanced/README.md

GitHub - aws/aws-sdk-java-v2: The official AWS SDK for Java - Version 2

 

그리고 모든 코드는 github에서 확인 가능하다.

https://github.com/weduls/dynamodb-enhanced