Glue의 DynamicFrame 대신 Glue에서 DataFrame을 사용하여 데이터 입력 출력

작업 내용

csv 파일을 parket으로 변환합니다.

※"Glue의 사용법적인 ①(GUI로 작업 실행)"(이후 ①이라고만 씁니다)와 같은 처리입니다. 데이터 입력 및 출력 부분을 DynamicFrame에서 DataFrame으로 변경합니다.

작업 이름

se2_job14

이전 준비

IAM 롤 등은 ①을 확인해 주십시오.

이번에 사용하는 샘플 로그 파일(19건)

csvlog.csv

deviceid,uuid,appid,country,year,month,day,hour
iphone,11111,001,JP,2017,12,14,12
android,11112,001,FR,2017,12,14,14
iphone,11113,009,FR,2017,12,16,21
iphone,11114,007,AUS,2017,12,17,18
other,11115,005,JP,2017,12,29,15
iphone,11116,001,JP,2017,12,15,11
pc,11118,001,FR,2017,12,01,01
pc,11117,009,FR,2017,12,02,18
iphone,11119,007,AUS,2017,11,21,14
other,11110,005,JP,2017,11,29,15
iphone,11121,001,JP,2017,11,11,12
android,11122,001,FR,2017,11,30,20
iphone,11123,009,FR,2017,11,14,14
iphone,11124,007,AUS,2017,12,17,14
iphone,11125,005,JP,2017,11,29,15
iphone,11126,001,JP,2017,12,19,08
android,11127,001,FR,2017,12,19,14
iphone,11128,009,FR,2017,12,09,04
iphone,11129,007,AUS,2017,11,30,14

S3에 배치

$ aws s3 ls s3://test-glue00/se2/in0/
2018-01-02 15:13:27          0 
2018-01-02 15:13:44        691 cvlog.csv

실행 코드

적당한 Glue 작업을 작성해, 완료된 작업의 내용의 코드를 이하의 코드로 덧쓰기
inferSchema로 데이터형을 Spark측에서 자동으로 유추한다. DataFrame으로 하고 있으므로 타입이 필요하기 때문에.

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

args = getResolvedOptions(sys.argv, ['TempDir','JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

df = spark.read.format("csv").option("header", "true").option("inferSchema", "true").load('s3://test-glue00/se2/in0/*.csv', header=True)

output='s3://test-glue00/se2/out14/'
codec='snappy'

df.write.mode("overwrite").parquet(output,compression=codec)
job.commit()

크롤러 생성 및 실행, 테이블 확인

출력 대상 S3 경로에 크롤러를 만들고 크롤러 실행

완성 된 테이블은 다음과 같습니다.

Athena 확인

이쪽도 부디

Glue 사용법 요약
htps : // 코 m / 피오 호 07 / ms / 32f76 아 16cbf49f9f712f