[AWS][Athena] Pyathena를 통한 Athena 쿼리

0. Overview

Pyathena 는 AWS Athena를 위한 DB API를 제공하는 Package이다.
해당 글은 외부 서버(On-Prem 등) 에서 Python 내 Pyathena 패키지를 활용하여 Athena 에 쿼리를 날리는 과정을 보여준다.

1. VPC Endpoint 생성

외부 서버에서 Athena Query 를 실행시키기 위해서는 VPC Endpoint를 이용하게 된다.

VPC Endpoint 는 VPC와 AWS 서비스들을 Private 하게 연결해주고 Client 는 VPC Endpoint 의 IP / DNS 등으로 접근할 수 있다.

인터페이스 VPC 엔드포인트를 사용하여 Amazon Athena에 연결 - Amazon Athena

 

VPC Endpoint 생성은 아래 글을 참고하여 생성한다.

https://docs.aws.amazon.com/vpc/latest/privatelink/create-interface-endpoint.html#create-interface-endpoint

Access an AWS service using an interface VPC endpoint - Amazon Virtual Private Cloud

 

 

1-1) Hosts 등록

추가적으로 Client 서버의 /etc/hosts에 Athena Endpoint의 dns를 등록해 주어야 한다.

아래는 등록 예시이다.

$ vim /etc/hosts

10.241.108.31	athena.ap-northeast-2.amazonaws.com

만약 Athena Endpoint의 설정에서 Private DNS names enabled 가 NO로 설정했다면 있다면 이 과정이 필요 없다.

2. Security Group 수정

athena VPC Endpoint 에 연결된 Security Group 내 Inbound Rule에 Source IP 를 등록해준다.
Port 제한이 필요하다면, 443(ssl) / 444 (jdbc)port를 사용하므로 해당 port로 제한하면 된다.

내부서버 (Ex. 같은 VPC내 EC2) 라면 필요 없는 과정이다.

 

3. Install Pyathena

$ pip install PyAthena

 

4. Set Default Credential

Option 2의 credential 을 를 복사하여 ~/.aws/credentials 경로에 credentials 파일을 생성해준다.
이 때, [default] credential 을 생성하면, 추후 pyathena를 통한 connect 생성 시 따로 access key, secret access key 등을 지정해주지 않아도 된다.

 

5. Execute Query to Athena

connection 생성 후 dataframe으로 바로 불러올 수 있도록 pandas의 real_sql_query 함수를 통해 쿼리를 실행시킬 수 있다.

from pyathena import connect
import pandas as pd

conn = connect( region_name = 'ap-northeast-2', 
                s3_staging_dir = "s3://path/here" )

df = pd.read_sql_query("SELECT * FROM test1.svc_day limit 2", conn)
df.show()

만약, default credential을 생성을 해주지 않았다면 connect를 맺을 때 access key, secret access key 등의 정보를 입력해야 한다.
4. 에서의 Option 2를 복사해여 ~/credential 파일에 저장했다 가정하고, 아래와 같이 파일을 읽어 credential 정보를 불러 올 수 있다.

from pyathena import connect
import pandas as pd

with open("~/credential") as f:
    lines = f.readlines()
    credential_dict = dict()
    for i in range(2,len(lines)):
        info = lines[i].split("=")
        credential_dict[info[0]]=info[1].strip("\n")

conn = connect( region_name = 'ap-northeast-2', 
               aws_access_key_id = credential_dict["aws_access_key_id"],
               aws_secret_access_key= credential_dict["aws_secret_access_key"],
               aws_session_token = credential_dict["aws_session_token"],
               s3_staging_dir = "s3://path/here" )

df = pd.read_sql_query("SELECT * FROM test1.svc_day limit 2", conn)
df.show()

Reference
https://pypi.org/project/pyathena/