Glueの使い方的な⑪(DynamicFrameでpartitionByが使えるようになった)

# 概要

従来DataFrameでしかできなかったWrite時のpartitionByに、DynamicFrameが対応しました。
https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-partitions.html

今回のやってみた内容は、"[Glueの使い方的な②(csvデータをパーティション分割したparquetに変換)](https://qiita.com/pioho07/items/523aec26ca5dc5bc9697)"と同じです。
csvデータをパーティション分割したparquetに変換します。

# パーティション分割するジョブを作る
## ジョブの内容

※"[Glueの使い方的な①(GUIでジョブ実行)](https://qiita.com/pioho07/items/c9ce1d0677777f974ffe)"(以後①とだけ書きます)と同様のcsvデータを使います

<b>"csvデータのタイムスタンプのカラムごとにパーティション分割してparquetで出力する"<b>

### ジョブ名
se2_job7

### クローラー名
se2_in0 (Glueの使い方的な②で作ったもの)
se2_out6

# 全体の流れ

* 前準備
* ジョブ作成と修正
* ジョブ実行と確認
* 出力データのクローラー作成、実行、Athenaで確認
* 別のカラムでパーティション分割

※①のGUIで作成したPySparｋスクリプトに最小限の変更を入れる形で進めます

# 前準備

### ソースデータ(19件)

※①と同じデータ

```csv: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
```
データの場所

※①と同じ場所

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

### S3のディレクトリ構成

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

```
$ aws s3 ls s3://test-glue00/se2/
                           PRE in0/
                           PRE out0/
                           PRE out1/
                           PRE script/
                           PRE tmp/
```

### 入力テーブルのクローラー

※①で作ったものを使います。

テーブルの情報は以下です。

![スクリーンショット 0030-01-03 15.45.21.png](https://qiita-image-store.s3.amazonaws.com/0/27932/4d12c905-75b9-aa34-3d01-609361d53157.png)

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

# ジョブ作成と修正

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

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

```python:se2_job7
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_in0", transformation_ctx = "datasource0"]
## @return: datasource0
## @inputs: []
datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "se2", table_name = "se2_in0", transformation_ctx = "datasource0")
## @type: ApplyMapping
## @args: [mapping = [("deviceid", "string", "deviceid", "string"), ("uuid", "long", "uuid", "long"), ("appid", "long", "appid", "long"), ("country", "string", "country", "string"), ("year", "long", "year", "long"), ("month", "long", "month", "long"), ("day", "long", "day", "long"), ("hour", "long", "hour", "long")], 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"), ("year", "long", "year", "long"), ("month", "long", "month", "long"), ("day", "long", "day", "long"), ("hour", "long", "hour", "long")], 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/out6"}, 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/out6"}, format = "parquet", transformation_ctx = "datasink4")
job.commit()
```

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

42行目の"datasink4"の行を以下のように修正

```python:
partitionby=['year','month','day','hour']
datasink4 = glueContext.write_dynamic_frame.from_options(frame = dropnullfields3, connection_type = "s3", connection_options = {"path": "s3://test-glue00/se2/out6", "partitionKeys": ['year','month','day','hour']}, format = "parquet", transformation_ctx = "datasink4")
```

TODO

# ジョブ実行と確認

### ジョブ実行

対象ジョブにチェックを入れ、ActionからRun jobをクリックしジョブ実行します
出力が指定したyaerやmonthでパーティション分割されている。

![スクリーンショット 0030-03-10 20.27.07.png](https://qiita-image-store.s3.amazonaws.com/0/27932/cebb299b-8bd5-22ad-3503-1e7e01d379cc.png)

### コマンドで確認

s3のparquetファイルを確認
ローカルにダウンロードし、parquet-toolsで内容確認

```shell:
# aws s3 cp s3://test-glue00/se2/out6/year=2017/month=11/day=14/hour=14/part-00000-c1945f99-ae0b-4d28-8642-49b87521b8da.snappy.parquet .
download: s3://test-glue00/se2/out6/year=2017/month=11/day=14/hour=14/part-00000-c1945f99-ae0b-4d28-8642-49b87521b8da.snappy.parquet to ./part-00000-c1945f99-ae0b-4d28-8642-49b87521b8da.snappy.parquet
# java -jar /root/parquet/parquet-mr/parquet-tools/target/parquet-tools-1.6.0rc7.jar head part-00000-c1945f99-ae0b-4d28-8642-49b87521b8da.snappy.parquet 
deviceid = iphone
uuid = 11123
appid = 9
country = FR

```

# 出力データのクローラー作成、実行、Athenaで確認

### se2_out6でクローラー作成

GlueのCrawlersをクリックし、"Add crawler"をクリック

![スクリーンショット 0030-03-10 20.38.50.png](https://qiita-image-store.s3.amazonaws.com/0/27932/e09349ac-48ac-6d14-b7cd-09c59cc6b625.png)

S3の出力パスを入力

![スクリーンショット 0030-03-10 20.39.29.png](https://qiita-image-store.s3.amazonaws.com/0/27932/34961d3b-13cc-915a-e0f0-326209aa937a.png)

そのまま"Next"をクリック

![スクリーンショット 0030-01-03 16.38.46.png](https://qiita-image-store.s3.amazonaws.com/0/27932/0799dcee-65a6-70a0-c791-ef11d2e3e3b5.png)

IAM roleに”test-glue"を選択

![スクリーンショット 0030-01-03 16.38.55.png](https://qiita-image-store.s3.amazonaws.com/0/27932/d6af42a8-ce2e-f6d3-668b-7462f345ea0c.png)

そのまま"Next"をクリック

![スクリーンショット 0030-01-03 16.39.03.png](https://qiita-image-store.s3.amazonaws.com/0/27932/79f168f9-496c-7c5d-1f74-bb66592d35e7.png)

Databaseを選択(今回はse2)
Prefixを入力(今回はse2_)

![スクリーンショット 0030-01-03 16.39.16.png](https://qiita-image-store.s3.amazonaws.com/0/27932/68a3ebdd-e4bb-bb55-a263-39a6b504e201.png)

### クローラー実行

クローラー実行後

![スクリーンショット 0030-03-10 20.46.07.png](https://qiita-image-store.s3.amazonaws.com/0/27932/8ac571a0-f636-d00b-9ee7-ee9f8e5d46b7.png)

スキーマも、yearやmonthなどで分割したパーティションを認識している

![スクリーンショット 0030-03-10 20.46.30.png](https://qiita-image-store.s3.amazonaws.com/0/27932/6ed9a34b-b26c-49ac-22b3-4d9eaeb60d00.png)


### Athenaから確認

左メニューからse2_out6のスキーマ情報確認、クエリ実行

![スクリーンショット 0030-03-10 20.48.35.png](https://qiita-image-store.s3.amazonaws.com/0/27932/156d58bd-6f40-4990-8a02-ce6b41b61cba.png)

件数も１９件で合っている

![スクリーンショット 0030-03-10 20.48.50.png](https://qiita-image-store.s3.amazonaws.com/0/27932/e106c027-3c3b-1e08-f472-61ae66cd3458.png)

## To Be Continue

よくありそうな変換処理ケースを今後書いていければと思います。

## こちらも是非

参考サンプル
https://github.com/awslabs/aws-big-data-blog/blob/master/aws-blog-spark-parquet-conversion/convert2parquet.py

Spark API
https://spark.apache.org/docs/preview/api/python/pyspark.sql.html
https://hackernoon.com/managing-spark-partitions-with-coalesce-and-repartition-4050c57ad5c4

Glueの使い方まとめ
https://qiita.com/pioho07/items/32f76a16cbf49f9f712f