[AWS S3] 임시테이블을 활용한 1억 4천만건 데이터 삭제 요청

https://girinprogram93.tistory.com/62

[CURSOR PAGING] 커서 페이징처리.

 

작업 환경 : java 1.8 / spring boot / gradle / jdbcTemplate / tomcat 9.06 / MySQL 

 

 

 

 

🍔 지난번의 커서 기반 페이징 S3 삭제 요청에 이어서 최종본이다.  ( 지금은 삭제된 데이터를 전수검사하고 있다. )

 

 

🥣 상황 : S3 스토리지에 삭제 대상인 특정 오브젝트가 1억 4천만건이 존재하며, 이를 삭제 해야한다. 

 

S3 스토리지의 오브젝트 key 경로는 운영 DB에 저장되어있으며, 

 

Key 경로를 가지고 있는 Row의 index(ID)가 일정하지 않고 ( 심할 경우 몇만 이상의 차이가 나는 ID 값들이 존재 ) 

(때문에 오프셋 방식보다 SELECT 속도가 빠른, 커서 페이징 방식의 select를 쓸 수 없었음.)

 

어쨌든,

S3는 다수의 오브젝트 삭제 요청을 할 경우 1회당 1000건의 오브젝트만 삭제 가능하도록 지원하기 때문에,  

 

SELECT 쿼리를 통해 데이터를 가져올 경우 점차 OFFSET 의 수가 늘어나게 되는데 이런 경우에는 SELECT 쿼리의 속도가 느려지는 단점이 있다. ( offset을 쓴다면 offset까지의 데이터를 다 SELECT 하고 LIMIT 범위 만큼만 잘라내는 방식이기 떄문에 ) 

 

따라서 다음과 같은 작업 계획으로 삭제 속도를 최대한 빠르게 늘리려고 함.

 

🥞 작업 계획 :

* 임시 테이블은 운영에 영향이 안가기 위해 개발 서버에 만들었다. 간단하게 구현하기 위해 jdbcTemplate으로 구현했고 접속 정보는 나눠두었다.

 

while ( breakFlag ) {

    1. tb_origin 에서 작업 시작 기준이 될 id 및 objectPath를 100만건 단위로 조회
    2. tb_temp에 insert --> 1000만건까지 등록한다.
    3. tb_temp의 id 및 path 인 데이터 조회 (select query) 는 1000단위로.  --> offset 단위가 0 ~ 1000만건으로, 작아지기 때문에 셀렉트 속도가 빨라짐.
    4. s3 삭제 요청 반복 (4 thread로 진행.) -> tb_temp의 1000만건을 모두 삭제요청하도록.
    5. tb_temp clear.

    if (bc_scene 의 작업 시작 기준이 될 scene_id 가 100만건 보다 작을 경우 || SELECT 한 값이 100만건보다 작을 경우.) {
        breakFlag = true;
    }
}

 

 

 

 

build.gradle 

plugins {
    id 'org.springframework.boot' version '2.7.1'
    id 'io.spring.dependency-management' version '1.0.11.RELEASE'
    id 'java'
    id 'war'
}

group = 'com.boot.aws'
version = ''
sourceCompatibility = '1.8'
apply plugin: 'war'

configurations {
    compileOnly {
        extendsFrom annotationProcessor
    }
}

repositories {
    mavenCentral()
    maven { url 'https://repo.spring.io/snapshot' }
    maven { url 'https://repo.spring.io/milestone' }
}

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-web'
    compileOnly 'org.projectlombok:lombok'
    annotationProcessor 'org.projectlombok:lombok'
    providedRuntime 'org.springframework.boot:spring-boot-starter-tomcat'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'

    runtimeOnly 'mysql:mysql-connector-java'

    implementation group: 'io.springfox', name: 'springfox-boot-starter', version: '3.0.0'
    implementation group: 'io.swagger', name: 'swagger-annotations', version: '1.6.2'
    implementation group: 'io.swagger', name: 'swagger-models', version: '1.6.2'

    // https://mvnrepository.com/artifact/com.amazonaws/aws-java-sdk-s3
    implementation group: 'com.amazonaws', name: 'aws-java-sdk-s3', version: '1.11.584'

    // mybatis
    implementation("org.mybatis.spring.boot:mybatis-spring-boot-starter:2.2.2")

}

 

 

logback-spring.xml 설정 : 

<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60 seconds">

    <!--
        적용 순서 :
        1. classpath 하위 logback-spring.xml
        2. .yml ( properties)
        3. 1, 2가 동시에 존재 할 경우 yml 적용 후, xml 적용됨.
    -->

    <!--<property name="LOG_FILE_NAME" value=""/>-->
    <!--<property name="ERR_LOG_FILE_NAME" value="err_log"/>-->


    <!--  코코스 로그 경로   -->
    <property name="LOG_PATH" value="/home/archive/demo/logs"/>

    <!-- 아카이브 로그  -->
    <!--<property name="LOG_PATH" value="/home/archive/demolog"/>-->
    <property name="LOG_PATTERN" value="%-5level %d{yy-MM-dd HH:mm:ss}[%thread] [%logger{0}:%line] - %msg%n"/>

    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <Pattern>${FILE_LOG_PATTERN}</Pattern>
        </encoder>
    </appender>

    <!-- Console Appender -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <pattern>${LOG_PATTERN}</pattern>
        </encoder>
    </appender>

    <!-- File Appender -->
    <appender name="demo" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <!-- 파일경로 설정 -->
<!--        <file>${LOG_PATH}/archive_demo.log</file>-->
        <file>${LOG_PATH}/demo.log</file>

        <!-- 출력패턴 설정-->
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <pattern>${LOG_PATTERN}</pattern>
        </encoder>

        <!-- Rolling 정책 -->
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!-- .gz,.zip 등을 넣으면 자동 일자별 로그파일 압축 -->
            <fileNamePattern>${LOG_PATH}/demo.%d{yyyy-MM-dd}_%i.log</fileNamePattern>
<!--            <fileNamePattern>${LOG_PATH}/archive_demo.%d{yyyy-MM-dd}_%i.log</fileNamePattern>-->
            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <!-- 파일당 최고 용량 kb, mb, gb -->
                <maxFileSize>100MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
            <!-- 일자별 로그파일 최대 보관주기(~일), 해당 설정일 이상된 파일은 자동으로 제거-->
            <maxHistory>120</maxHistory>
            <!--<MinIndex>1</MinIndex>
            <MaxIndex>10</MaxIndex>-->
        </rollingPolicy>
    </appender>

    <!-- 에러의 경우 파일에 로그 처리 -->
    <appender name="ERROR" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${LOG_PATH}/error.log</file>
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <pattern>${LOG_PATTERN}</pattern>
        </encoder>
        <!-- Rolling 정책 -->
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!-- .gz,.zip 등을 넣으면 자동 일자별 로그파일 압축 -->
<!--            <fileNamePattern>${LOG_PATH}/error_archive.%d{yyyy-MM-dd}_%i.log</fileNamePattern>-->
            <fileNamePattern>${LOG_PATH}/error.%d{yyyy-MM-dd}_%i.log</fileNamePattern>
            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <!-- 파일당 최고 용량 kb, mb, gb -->
                <maxFileSize>100MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
            <!-- 일자별 로그파일 최대 보관주기(~일), 해당 설정일 이상된 파일은 자동으로 제거-->
            <maxHistory>120</maxHistory>
        </rollingPolicy>
    </appender>

    <!-- 삭제 검증 로직 용 로그 -->
    <appender name="delChk" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <!-- 파일경로 설정 -->
        <!--        <file>${LOG_PATH}/archive_demo.log</file>-->
        <file>${LOG_PATH}/del_chk.log</file>

        <!-- 출력패턴 설정-->
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
            <pattern>${LOG_PATTERN}</pattern>
        </encoder>

        <!-- Rolling 정책 -->
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <!-- .gz,.zip 등을 넣으면 자동 일자별 로그파일 압축 -->
            <fileNamePattern>${LOG_PATH}/cocos_del_chk.%d{yyyy-MM-dd}_%i.log</fileNamePattern>
            <!--            <fileNamePattern>${LOG_PATH}/archive_demo.%d{yyyy-MM-dd}_%i.log</fileNamePattern>-->
            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <!-- 파일당 최고 용량 kb, mb, gb -->
                <maxFileSize>100MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
            <maxHistory>120</maxHistory>
        </rollingPolicy>
    </appender>


    <logger name="com.error" additivity="false">
        <appender-ref ref="ERROR"/>
    </logger>

    <logger name="com.demo" additivity="false">
        <appender-ref ref="demo"/>
    </logger>

    <logger name="com.delChk" additivity="false">
        <appender-ref ref="delChk"/>
    </logger>

    <!-- root레벨 설정 -->
    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
        <appender-ref ref="ERROR"/>
    </root>

</configuration>

 

 

 

 

JdbcTemplate 설정 : 

## application.yml
## jdbc를 각각 나눠서 쓸 수 있도록 설정

server:
  port: 9200
  servlet:
    context-path: /awsdemo/

spring:
  datasource-dev: # tb_temp dev
    driverClassName: com.mysql.jdbc.Driver
    jdbcUrl: jdbc:mysql://127.255.0.11:3306/dcmsdb?serverTimezone=UTC&characterEncoding=UTF-8
    username: dev
    password: samplePassword 
  datasource-real: # tb_origin real
    driverClassName: com.mysql.jdbc.Driver
    jdbcUrl: jdbc:mysql://127.255.0.22:3306/dcmsdb?serverTimezone=UTC&characterEncoding=UTF-8
    username: orign
    password: sampleOriginPassword

  h2.console.enabled: true
  jpa:
    show-sql: true
    hibernate:
      # create로 설정하면 실행 시, 기존 테이블을 삭제하고 새로 생성하게 됨.
      #ddl-auto: validate
      ddl-auto: none
    defer-datasource-initialization: true
  output:
    ansi:
      enabled: always
logging:
  level:
    org.hibernate.type: trace

 

 

 

PersistentConfig.java 

package com.boot.aws.awsdemo;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.jdbc.DataSourceBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.jdbc.core.JdbcTemplate;

import javax.sql.DataSource;

@Configuration
public class PersistentConfig {

    @Primary
    @Bean(name = "dev")
    @ConfigurationProperties(prefix="spring.datasource-dev")
    public DataSource dsSlave0() {
        return DataSourceBuilder.create().build();
    }


    @Bean(name = "real")
    @ConfigurationProperties(prefix="spring.datasource-real")
    public DataSource dsSlave1() {
        return DataSourceBuilder.create().build();
    }

    @Bean(name = "devJdbcTemplate")
    @Autowired
    public JdbcTemplate devJdbcTemplate(@Qualifier("dev") DataSource dsSlave) {
        return new JdbcTemplate(dsSlave);
    }

    @Bean(name = "realJdbcTemplate")
    @Autowired
    public JdbcTemplate realJdbcTemplate(@Qualifier("real") DataSource dsSlave) {
        return new JdbcTemplate(dsSlave);
    }

}

 

 

 

 

AwsDeleteDemo.java 

package com.boot.aws.awsdemo;

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.amazonaws.services.s3.model.DeleteObjectsRequest;
import io.swagger.annotations.ApiOperation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Controller;
import org.springframework.util.ObjectUtils;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;

import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Controller
@RequestMapping("/aws2")
@EnableAutoConfiguration
public class AwsDeleteDemo {

    private final Logger log_demo = LoggerFactory.getLogger("com.demo");
    private final Logger log_error = LoggerFactory.getLogger("com.error");

    @Autowired
    @Qualifier("devJdbcTemplate")
    private JdbcTemplate devTemplate;

    @Autowired
    @Qualifier("realJdbcTemplate")
    private JdbcTemplate realTemplate;

    static int workCount = 0;
    static int offset = 0;
    static int threadOffset = -1000;

    @ApiOperation(value = "aws s3 object delete api", notes = "")
    @RequestMapping(value = "/del/{id}", method = RequestMethod.GET, produces = MediaType.APPLICATION_JSON_VALUE)
    public ResponseEntity<?> s3DeleteObjList2(
            @RequestParam("id") Integer id
    ) {

        try {

            log_demo.info("오프셋 출발점 : " + id);

            boolean breakFlag = false;

            while ( !breakFlag ) {
                int tempCount = 0;

                for (int i = 0 ; i < 10; i ++ ) {

                    List<Dto> ids = realTemplate.query(
                            "SELECT id, path from tb_origin limit 1000000 offset " + id
                            , new BeanPropertyRowMapper<Dto>(Dto.class));

                    tempCount+= ids.size();

                    log_demo.info(" 임시테이블 데이터 등록 개수 : " + tempCount );

                    // 2.
                    // 개발 서버 임시 테이블에 id insert
                    devTemplate.batchUpdate("INSERT INTO tb_temp (id, path) values (?,?) ",
                            new BatchPreparedStatementSetter() {
                                @Override
                                public void setValues(PreparedStatement ps, int i) throws SQLException {
                                    ps.setInt(1, ids.get(i).getId());
                                    ps.setString(2, ids.get(i).getPath());
                                }
                                @Override
                                public int getBatchSize() {
                                    return ids.size();
                                }
                            }
                    );

                    log_demo.info("개발 서버 임시 테이블에 " + tempCount + " 건이 등록되었습니다.");

                    // 총 삭제 카운트가 몇개 인지 알고 있다면, 다음 개수가 끝나면 멈추도록 한다.
                    if (id > 154553000) {
                        log_demo.info(" offset : " + id + " 총 타겟 개수 : " + 154553000 + " 종료 플래그 true로 변경합니다.");
                        breakFlag = true;
                    }


                    id += 1000000;
                    log_demo.info(" 코코스 다음 select offset : " + id);
                }

                

                offset = -1000;

                int AllCount = 154553000;
                
                // 4개의 쓰레드로 동시에 작업하도록 했다.
                ExecutorService executorService = Executors.newFixedThreadPool(4);
                while(offset < AllCount){
                    for (int i = 0; i <4; i ++ ) {
                        Thread.sleep(100);
                        executorService.submit(() -> {

                            offset+=1000;
                            boolean flag = ThreadRunner(AllCount, offset);

                        });
                    }
                    if (executorService.isTerminated() || executorService.isShutdown()) {
                        log_demo.info(" 쓰레드 종료.");
                    }
                }

                // 5. 임시 테이블의 데이터를 삭제
                int tempRows = devTemplate.update("DELETE FROM tb_temp ");
                log_demo.info(" 임시테이블의 데이터 [ " + tempRows + " ] 건이 삭제되었습니다. " );

                if (!breakFlag) {
                    log_demo.info(" 다음 작업 진행... id target : " + id);
                }
            }

        } catch (Exception e) {
            ByteArrayOutputStream out = new ByteArrayOutputStream();
            PrintStream pinrtStream = new PrintStream(out);
            e.printStackTrace(pinrtStream);
            log_error.error(out.toString());
            return new ResponseEntity<>( HttpStatus.INTERNAL_SERVER_ERROR);
        } finally {
            log_demo.info("작업이 종료되었습니다.");
        }

        return new ResponseEntity<String>("SUCCESS!!!", HttpStatus.OK);
    }


    public boolean ThreadRunner(int allCount, int offset) {

        String endpoint = "https://aws.endpoint_sample";
        String region = "ap-northeast-2";
        String accesskey = "accessKey++";
        String secretkey = "secretKey+++";
        String bucketname = "bucketName";

        AWSCredentials credentials = new BasicAWSCredentials(accesskey, secretkey);

        // Build the Client
        AmazonS3 s3 = AmazonS3ClientBuilder.standard()
                .withCredentials(new AWSStaticCredentialsProvider(credentials))
                .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpoint ,region))
                .withPathStyleAccessEnabled(true)
                .build();

        log_demo.info(" target all count : " + allCount);

        int limit = 1000;

        try {
            log_demo.info("runner! ");
            List<DeleteObjectsRequest.KeyVersion> Arrays = new ArrayList<>();

            // tb-_temp 의 id , path를 조회한다 (1000개 단위 )
            List<Dto> target = devTemplate.query("select id " +
                    ",path" +
                    " from tb_temp as s" +
                    " limit 1000 OFFSET " + offset, new BeanPropertyRowMapper<Dto>(Dto.class));

            log_demo.info(" offset : " + offset);

            for (int i =0; i < target.size(); i ++) {
                String temp = "public/"+target.get(i).getPath();
                log_demo.info(target.get(i).getId() + " / " + temp);
                Arrays.add(new DeleteObjectsRequest.KeyVersion(temp));
            }

			// 실제 삭제 요청.
            DeleteObjectsRequest multiObjectDeleteRequest = new DeleteObjectsRequest(bucketname).withKeys(Arrays);
            long beforeTime = System.currentTimeMillis();
            s3.deleteObjects(multiObjectDeleteRequest);
            long afterTime = System.currentTimeMillis();
            long time_chker = (afterTime - beforeTime);

            // 삭제 요청 이후 리턴 받는 시간을 체크한다.
            log_demo.info(" delete required time chk : " + time_chker);

            workCount += target.size();
            log_demo.info("전체 작업 진행 개수 : " + workCount);

        } catch ( Exception r){
            ByteArrayOutputStream out = new ByteArrayOutputStream();
            PrintStream printStream = new PrintStream(out);
            r.printStackTrace(printStream);
            log_error.error(out.toString());
            return false;
        }

        return true;
    }


}

 

임시 테이블에 등록되는 과정 기록 : 

 

 

 

 

 

 

 

삭제 한 로그 기록 : id / path 

 

 

 

 

삭제한 오브젝트를 일일이 확인 하기가 귀찮다면, 

 

https://s3browser.com/

S3 Browser - Amazon S3 Client for Windows. User Interface for Amazon S3. S3 Bucket Explorer.

 

 

S3 Object를 디렉토리별로 나눠서 보여주는 툴.