パーティション分割するジョブを作る

ジョブの内容

元データに"2018/08/26 12:00:01"みたいなタイムスタンプが入ってるデータを、パーティションによるディレクトリ構成にしてデータ配置とフォーマットなど変換する

※"Glueの使い方的な①(GUIでジョブ実行)"(以後①とだけ書きます)と同じ内容のデータで、ディレクトリyyyy/mmに分割してファイルを配置しています。

ジョブ名

se2_job12

クローラー名

se2_in10
se2_out12

全体の流れ

前準備
入力データのクローラー作成と実行
ジョブの作成と修正
ジョブの実行と確認

前準備

ソースデータ(19件)

csvlog-20171111.csv-csvlog-20171229.csv

$ ls
cvlog-20171111.csv  cvlog-20171201.csv  cvlog-20171216.csv
cvlog-20171114.csv  cvlog-20171202.csv  cvlog-20171217.csv
cvlog-20171121.csv  cvlog-20171209.csv  cvlog-20171219.csv
cvlog-20171129.csv  cvlog-20171214.csv  cvlog-20171229.csv
cvlog-20171130.csv  cvlog-20171215.csv

ヘッダ情報こんな感じ

ヘッダ情報

$ cat * | head
deviceid,uuid,appid,country,timestamp
iphone,11121,001,JP,2017/11/11 12:00:01
deviceid,uuid,appid,country,timestamp
iphone,11123,009,FR,2017/11/14 14:00:01
deviceid,uuid,appid,country,timestamp
iphone,11119,007,AUS,2017/11/21 14:00:01
deviceid,uuid,appid,country,timestamp
other,11110,005,JP,2017/11/29 15:55:01
iphone,11125,005,JP,2017/11/29 15:00:01
deviceid,uuid,appid,country,timestamp

データの内容

csvlog-20171111.csv-csvlog-20171229.csv

$ cat * | grep -v deviceid 
iphone,11121,001,JP,2017/11/11 12:00:01
iphone,11123,009,FR,2017/11/14 14:00:01
iphone,11119,007,AUS,2017/11/21 14:00:01
other,11110,005,JP,2017/11/29 15:55:01
iphone,11125,005,JP,2017/11/29 15:00:01
android,11122,001,FR,2017/11/30 20:00:01
iphone,11129,007,AUS,2017/11/30 14:33:01
pc,11118,001,FR,2017/12/01 01:00:01
pc,11117,009,FR,2017/12/02 18:00:01
iphone,11128,009,FR,2017/12/09 04:11:01
iphone,11111,001,JP,2017/12/14 12:00:01
android,11112,001,FR,2017/12/14 14:00:01
iphone,11116,001,JP,2017/12/15 11:00:01
iphone,11113,009,FR,2017/12/16 21:00:01
iphone,11114,007,AUS,2017/12/17 18:00:01
iphone,11124,007,AUS,2017/12/17 14:00:01
iphone,11126,001,JP,2017/12/19 08:00:01
android,11127,001,FR,2017/12/19 14:10:01
other,11115,005,JP,2017/12/29 15:00:01

件数19件

件数

$ cat * | grep -v deviceid |wc -l
      19

S3にデータをアップロード

$ aws s3 ls s3://test-glue00/se2/in10/ --recursive
2018-08-26 09:48:30          0 se2/in10/
2018-08-26 09:49:17          0 se2/in10/2017/
2018-08-26 09:49:24          0 se2/in10/2017/11/
2018-08-26 09:49:41         78 se2/in10/2017/11/cvlog-20171111.csv
2018-08-26 09:49:41         78 se2/in10/2017/11/cvlog-20171114.csv
2018-08-26 09:49:41         79 se2/in10/2017/11/cvlog-20171121.csv
2018-08-26 09:49:42        117 se2/in10/2017/11/cvlog-20171129.csv
2018-08-26 09:49:42        120 se2/in10/2017/11/cvlog-20171130.csv
2018-08-26 09:49:26          0 se2/in10/2017/12/
2018-08-26 09:49:57         74 se2/in10/2017/12/cvlog-20171201.csv
2018-08-26 09:49:58         74 se2/in10/2017/12/cvlog-20171202.csv
2018-08-26 09:49:58         78 se2/in10/2017/12/cvlog-20171209.csv
2018-08-26 09:49:58        119 se2/in10/2017/12/cvlog-20171214.csv
2018-08-26 09:49:58         78 se2/in10/2017/12/cvlog-20171215.csv
2018-08-26 09:49:58         78 se2/in10/2017/12/cvlog-20171216.csv
2018-08-26 09:49:59        120 se2/in10/2017/12/cvlog-20171217.csv
2018-08-26 09:49:59        119 se2/in10/2017/12/cvlog-20171219.csv
2018-08-26 09:49:59         77 se2/in10/2017/12/cvlog-20171229.csv

S3のディレクトリ構成

Glueジョブの入力データは"in10"ディレクトリ配下、出力は"out12"ディレクトリ配下

$ aws s3 ls s3://test-glue00/se2/
                           PRE in10/
                           PRE out12/
                           PRE script/
                           PRE tmp/

入力データのクローラー作成と実行

入力データのクローラー作成

AWSマネージメントコンソールから、Glueをクリックし、画面左側メニューの"クローラー"をクリックし、"クローラーの追加"をクリック

"クローラーの名前"を入力し[次へ]をクリック

"Choose a data store"にS3を選択し、"インクルードパス"にS3にあるソースデータのパス入力し[次へ]をクリック。その次も[次へ]をクリックし、その次は適切なIAMロールを選択し[次へ]をクリックし、その次も[次へ]をクリックする

"データベース"に"se2"を入力し、"プレフィックス"に"se2_"を入力し[次へ]をクリック、次の画面で[完了]をクリックする

クローラー実行

作成されたクローラー"se2_in10"にチェックを入れ[クローラの実行]をクリックする

作成されたテーブル情報を確認

Athenaでも確認

このあと、ジョブ作成とPySparkスクリプト修正、出力データのクローラー作成を行っていきます

ジョブ作成と修正

ジョブの作成

①と同じ手順でGUIのみの操作でse2_job12ジョブを作成
この段階では①とほぼ同じ内容のジョブです
コードは以下になります。

処理内容は"S3の指定した場所に配置したcsvデータを指定した場所にparquetとして出力する"です。

se2_job12

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

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
## @type: DataSource
## @args: [database = "se2", table_name = "se2_in10", transformation_ctx = "datasource0"]
## @return: datasource0
## @inputs: []
datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "se2", table_name = "se2_in10", transformation_ctx = "datasource0")
## @type: ApplyMapping
## @args: [mapping = [("deviceid", "string", "deviceid", "string"), ("uuid", "long", "uuid", "long"), ("appid", "long", "appid", "long"), ("country", "string", "country", "string"), ("timestamp", "string", "timestamp", "string")], transformation_ctx = "applymapping1"]
## @return: applymapping1
## @inputs: [frame = datasource0]
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("deviceid", "string", "deviceid", "string"), ("uuid", "long", "uuid", "long"), ("appid", "long", "appid", "long"), ("country", "string", "country", "string"), ("timestamp", "string", "timestamp", "string")], transformation_ctx = "applymapping1")
## @type: ResolveChoice
## @args: [choice = "make_struct", transformation_ctx = "resolvechoice2"]
## @return: resolvechoice2
## @inputs: [frame = applymapping1]
resolvechoice2 = ResolveChoice.apply(frame = applymapping1, choice = "make_struct", transformation_ctx = "resolvechoice2")
## @type: DropNullFields
## @args: [transformation_ctx = "dropnullfields3"]
## @return: dropnullfields3
## @inputs: [frame = resolvechoice2]
dropnullfields3 = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = "dropnullfields3")
## @type: DataSink
## @args: [connection_type = "s3", connection_options = {"path": "s3://test-glue00/se2/out12"}, format = "parquet", transformation_ctx = "datasink4"]
## @return: datasink4
## @inputs: [frame = dropnullfields3]
datasink4 = glueContext.write_dynamic_frame.from_options(frame = dropnullfields3, connection_type = "s3", connection_options = {"path": "s3://test-glue00/se2/out12"}, format = "parquet", transformation_ctx = "datasink4")
job.commit()

ジョブの修正

以下の部分を修正します。

冒頭のimportに以下を追加

from pyspark.sql.functions import to_timestamp
from pyspark.sql.functions import dayofmonth
from pyspark.sql.functions import month
from pyspark.sql.functions import unix_timestamp
from pyspark.sql.functions import year

35行目の"dropnullfields3"の後に以下を追加

df = dropnullfields3.toDF()
df = df.withColumn("timestamp2", to_timestamp(df.timestamp, 'yyyy/MM/dd HH:mm:ss'))
df = df.withColumn("year", year(df.timestamp2)).withColumn("month", month(df.timestamp2)).withColumn("day", dayofmonth(df.timestamp2))
partitionby=['year','month','day']
output='s3://test-glue00/se2/out12/'
codec='snappy'
df.repartition(*partitionby).write.partitionBy(partitionby).mode("append").parquet(output,compression=codec)

toDF：DynamicFrameをDataFrameに変換
write：DataFrameのデータを外部に保存。jdbc, parquet, json, orc, text, saveAsTable
parquetのcompression：none, snappy, gzip, and, lzoから選べる
partitionBy：Hiveパーティションのようにカラム=バリュー形式でパーティション化されたディレクトリにデータを保存
mode：ファイルやテーブルが既に存在してる場合の振る舞い。overwrite,append,ignore,error(デフォ)
repartition(numPartitions, *cols)[source]：パーティションの再配置、カッコ内はパーティションする単位を数字かカラムで選ぶ、カラムが優先

DataFrameのwrite()で出力するため、最後の方のsink処理をコメントアウト

#datasink4 = glueContext.write_dynamic_frame.from_options(frame = dropnullfields3, connection_type = "s3", connection_options = {"path": "s3://test-glue00/se2/out1"}, format = "parquet", transformation_ctx = "datasink4")

修正したコード

se2_job12_update

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
from datetime import datetime
from pyspark.sql.functions import to_timestamp
from pyspark.sql.functions import dayofmonth
from pyspark.sql.functions import month
from pyspark.sql.functions import unix_timestamp
from pyspark.sql.functions import year

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
## @type: DataSource
## @args: [database = "se2", table_name = "se2_in10", transformation_ctx = "datasource0"]
## @return: datasource0
## @inputs: []
datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "se2", table_name = "se2_in10", transformation_ctx = "datasource0")
## @type: ApplyMapping
## @args: [mapping = [("deviceid", "string", "deviceid", "string"), ("uuid", "long", "uuid", "long"), ("appid", "long", "appid", "long"), ("country", "string", "country", "string"), ("timestamp", "string", "timestamp", "string")], transformation_ctx = "applymapping1"]
## @return: applymapping1
## @inputs: [frame = datasource0]
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("deviceid", "string", "deviceid", "string"), ("uuid", "long", "uuid", "long"), ("appid", "long", "appid", "long"), ("country", "string", "country", "string"), ("timestamp", "string", "timestamp", "string")], transformation_ctx = "applymapping1")
## @type: ResolveChoice
## @args: [choice = "make_struct", transformation_ctx = "resolvechoice2"]
## @return: resolvechoice2
## @inputs: [frame = applymapping1]
resolvechoice2 = ResolveChoice.apply(frame = applymapping1, choice = "make_struct", transformation_ctx = "resolvechoice2")
## @type: DropNullFields
## @args: [transformation_ctx = "dropnullfields3"]
## @return: dropnullfields3
## @inputs: [frame = resolvechoice2]
dropnullfields3 = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = "dropnullfields3")
## @type: DataSink
## @args: [connection_type = "s3", connection_options = {"path": "s3://test-glue00/se2/out12"}, format = "parquet", transformation_ctx = "datasink4"]
## @return: datasink4
## @inputs: [frame = dropnullfields3]
df = dropnullfields3.toDF()
df = df.withColumn("timestamp2", to_timestamp(df.timestamp, 'yyyy/MM/dd HH:mm:ss'))
df = df.withColumn("year", year(df.timestamp2)).withColumn("month", month(df.timestamp2)).withColumn("day", dayofmonth(df.timestamp2))
partitionby=['year','month','day']
output='s3://test-glue00/se2/out12/'
codec='snappy'
df.repartition(*partitionby).write.partitionBy(partitionby).mode("overwrite").parquet(output,compression=codec)

#datasink4 = glueContext.write_dynamic_frame.from_options(frame = dropnullfields3, connection_type = "s3", connection_options = {"path": "s3://test-glue00/se2/out12"}, format = "parquet", transformation_ctx = "datasink4")
job.commit()