AWS Athena의 동시 질의를 위한 하우징 스크립트

다음은 셸 스크립트로 이루어집니다.

Athena에 질의를 제출합니다.

S3에서 출력 결과의 CSV를 다운로드합니다.

또 Athena의workspace 및database,table 등의 설치를 완료했다고 전제했다.

디렉토리 구조

Docker 컨테이너 내에서 실행되는 디렉터리는 다음과 같습니다.

.
├── docker-compose.yml
├── Dockerfile
├── queries
├── run.sh
└── csv

run.sh에서 Athena에 던진 조회를 위한 SQL 파일을 설정합니다.queries는 CSV 파일의 저장소 폴더입니다.

차리다

Athena에게 던진 SQL

csv에 SQL 파일을 배치하고 Athena에 던지는 조회를 작성합니다.Athena에서는 파티션이 대개 날짜에 의해 차단되므로 WERE 문을 변수로 대체합니다.
./queries/my_tables.sql

SELECT * FROM my_database.my_table WHERE date = :DATE_CONDITION

조개각본

셸 스크립트를 실행합니다.다음은 완성된 것이다.
run.sh

#!/usr/bin/env bash
set -eu

# Athenaのworkspace及びdatabase
WORK_GROUP=myapp
DATABASE=my_database

# UNIX_TIMESTAMPをフォーマットした文字列(yyyy-mm-dd)
CURRENT_DATE=$(date --date=@${UNIX_TIMESTAMP} +%Y-%m-%d)

# Athenaのクエリ結果(CSV)を出力するパス
QUERY_RESULT_PATH="s3://${S3_BUCKET}/my_database/${UNIX_TIMESTAMP}"

# Athenaのクエリが完了するまで待機する
wait_until_query_done() {
  QUERY_EXECUTION_ID=$1
  while :
  do
    QUERY_STATUS=$(aws athena get-query-execution \
      --query-execution-id ${QUERY_EXECUTION_ID} \
      | jq -r .QueryExecution.Status.State)
    echo "status of query which is executing on aws athena is $QUERY_STATUS"
    if [[ ${QUERY_STATUS} = "SUCCEEDED" ]] || \
        [[ ${QUERY_STATUS} = "FAILED" ]] || \
        [[ ${QUERY_STATUS} = "CANCELLED" ]]; then
      break;
    fi
    sleep 1
  done
}

# Athenaにクエリを投げる
# 返り値: AthenaのQUERY_EXECUTION_ID
start_query_on_athena() {
  TABLE=$1
  QUERY=$(sed -e "s/:DATE_CONDITION/\'${CURRENT_DATE}\'/g" < ./queries/${TABLE}.sql)

  QUERY_EXEC_ID=$(aws athena start-query-execution \
    --query-string "${QUERY}" \
    --query-execution-context Database=${DATABASE} \
    --result-configuration OutputLocation=${QUERY_RESULT_PATH}/${TABLE} \
    --work-group ${WORK_GROUP} \
    | jq -r '.QueryExecutionId')

  echo ${QUERY_EXEC_ID}
}

# Athenaにクエリを打って結果のCSVファイルをダウンロードする
download_files_from_s3() {
  TABLE=$1
  QUERY_EXEC_ID=$(start_query_on_athena $TABLE)
  echo "query_execution_id of query which is executing on aws athena is ${QUERY_EXEC_ID}"
  wait_until_query_done ${QUERY_EXEC_ID}
  aws s3 cp ${QUERY_RESULT_PATH}/${TABLE}/${QUERY_EXEC_ID}.csv /tmp/csv/${TABLE}.csv
}

download_files_from_s3 my_table

다음은 처리 개요입니다.

./queries에서 조회를 읽습니다.이 때./queries/*.sql 변수를 명령으로 바꿉니다.

읽은 검색어를 Athena에 버리고 처리가 끝날 때까지 기다립니다.

처리가 완료되면 S3에서 실행 결과의 CSV 파일을 다운로드합니다.

Dockerfile

sed.준비된 조회, 케이스 스크립트, 필요한 명령행 도구를 미리 설치합니다.또한 스크립트가 AWS 서비스에 의존하기 때문에 기본 이미지는 Dockerfile로 설정됩니다.
Dockerfile

FROM amazonlinux:2

RUN yum update -y
RUN yum install -y sudo date jq
RUN curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip" && \
        unzip awscliv2.zip && \
        sudo ./aws/install

COPY ./queries ./queries
COPY ./run.sh ./run.sh

ENV AWS_DEFAULT_REGION ap-northeast-1
ENV TZ Asia/Tokyo

ENTRYPOINT ["sh", "run.sh"]

docker-compose.yml

amazonlinux.docker-compose.yml에 다운로드된 CSV 파일의 저장소 폴더를 지정합니다.또한 환경 변수에 다음을 지정합니다.

S3_BUCKET: CSV 출력 대상 S3 배럴

UNIX_TIMESTAMP:Athena에 질의를 보낼 때 지정한 날짜 분할 조건

AWS...: AWS의 살결

docker-compose.yml

version: '3'
services:
  download-athena-result:
      build:
        context: ./
        dockerfile: ./Dockerfile
      image: download-athena-result
      environment:
        - S3_BUCKET
        - UNIX_TIMESTAMP
        - AWS_ACCESS_KEY_ID
        - AWS_SECRET_ACCESS_KEY
        - AWS_SESSION_TOKEN
      volumes: ./csv:/tmp/csv

실행 프로그램

환경 변수를 설정하고 실행합니다.

export S3_BUCKET=my-bucket
export UNIX_TIMESTAMP=xxx
export AWS_ACCESS_KEY_ID=xxx
export AWS_SECRET_ACCESS_KEY=xxx
export AWS_SESSION_TOKEN=xxx

docker-compose up

NOTE

조회가 간단하고 손쉽게 처리될 때는 좋지만, 조금 더 신경 쓰는 일을 하거나 병행 실행하고 싶다면 SDK를 솔직하게 사용해 다른 언어로 대응하는 것이 좋다.

끝맺다