# DB 定義
[cloudshell-user@XXXXXXXX ~]$ cat << EOS > database-definition.json
> {
>   "Name": "syslog",
>   "Description": "analyze syslogs."
> }
> EOS

# DB 作成
[cloudshell-user@XXXXXXXX ~]$ aws glue create-database --database-input file://database-definition.json

#確認
[cloudshell-user@XXXXXXXX ~]$ aws glue get-database --name syslog

#カスタムパターン定義の作成
#カラム名は分かれば何でもOK
[cloudshell-user@XXXXXXXX ~]$ cat << EOS > classifier-grok.json
> {
>   "Classification": "custom-log",
>   "Name": "audit-classifier",
>   "GrokPattern": "^%{DATA:type} %{DATA:msg} %{DATA:pid} %{DATA:uid} %{DATA:auid} %{DATA:ses} %{DATA: subj} %{DATA:msg2}$"
> }
> EOS

#作成
[cloudshell-user@XXXXXXXX ~]$ aws glue create-classifier --grok-classifier file://classifier-grok.json

#確認
[cloudshell-user@XXXXXXXX ~]$ aws glue get-classifier --name audit-classifier

# crawler の作成
[cloudshell-user@XXXXXXXX ~]$ CRAWLER_ROLE=AWSGlueServiceRole
[cloudshell-user@XXXXXXXX ~]$ DATABASE_NAME=syslog
[cloudshell-user@XXXXXXXX ~]$ CLASSIFIER=audit-classifier
[cloudshell-user@XXXXXXXX ~]$ S3_BUCKET_NAME='バケット名'
[cloudshell-user@XXXXXXXX ~]$ cat << EOS > audit.json
> {
>     "S3Targets": [
>         {
>             "Path": "s3://${S3_BUCKET_NAME}/audit",
>             "Exclusions": []
>         }
>     ],
>     "JdbcTargets": [],
>     "DynamoDBTargets": [],
>     "CatalogTargets": []
> }
> EOS

# crawler の作成
[cloudshell-user@XXXXXXXX ~]$ aws glue create-crawler --name audit-crawler --role $CRAWLER_ROLE --database-name $DATABASE_NAME --description "convert audit" --classifiers $CLASSIFIER --targets file://audit.json --tags Name=audit,Creation=cli

#確認
[cloudshell-user@XXXXXXXX ~]$ aws glue list-crawlers --query "CrawlerNames[?contains(@, 'audit')]"
[cloudshell-user@XXXXXXXX ~]$ aws glue get-crawler --name audit-crawler --query "Crawler.{NAME: Name, STATE: State, TIME: CrawlElapsedTime}"

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 pyspark.sql.functions import lit
from pyspark.sql.types import DateType
from pyspark.sql.functions import col
from pyspark.sql.functions import from_utc_timestamp
from awsglue.dynamicframe import DynamicFrame
from pyspark.sql.functions import date_format
from pyspark.sql.functions import split


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

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

#データカタログをソースに指定
datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "syslog", table_name = "audit", transformation_ctx = "datasource0")

#以下設定だと2022/5/15/12のデータをソースに指定することができる(一部のデータのみ読みたい際に利用可能)
#datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "syslog", table_name = "audit", transformation_ctx = "datasource0",push_down_predicate ="(year == '2022' and month == '05' and day == '15' and hour == '12')")


# string で各 field を取得
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("type", "string", "type", "string"), ("msg", "string", "msg", "string"), ("pid", "string", "pid", "string"), ("uid", "string", "uid", "string"), ("auid", "string", "auid", "string"), ("ses", "string", "ses", "string"), (" subj", "string", " subj", "string"), ("msg2", "string", "msg2", "string"), ("year", "string", "year", "string"), ("month", "string", "month", "string"), ("day", "string", "day", "string"), ("hour", "string", "hour", "string")], transformation_ctx = "applymapping1")

# dynamic frame → data frame への変換
df = applymapping1.toDF()

#--- audit 特有の変換
# msg から unix タイムを取得する
df = df.withColumn("msg", split(col("msg"), "\\(").getItem(1))
df = df.withColumn("date", split(col("msg"),":").getItem(0))
df = df.withColumn("date", split(col("date"),"\\.").getItem(0))

# date の型変換の実施
df = df.withColumn("timestamp", col("date").cast("Integer"))
df = df.withColumn("timestamp", col("timestamp").cast("Timestamp"))
df = df.withColumn("timestamp", from_utc_timestamp(col("timestamp"), 'JST'))

# split で=を分割
df = df.withColumn("type", split(col("type"), "=").getItem(1))
df = df.withColumn("pid", split(col("pid"), "=").getItem(1))
df = df.withColumn("uid", split(col("uid"), "=").getItem(1))
df = df.withColumn("auid", split(col("auid"), "=").getItem(1))
df = df.withColumn("ses", split(col("ses"), "=").getItem(1))
df = df.withColumn("msg2", split(col("msg2"), "=").getItem(1))
#---- audit 特有の変換

#データが分散されて複数のファイルで出力されてしまうため coalesce メソッドでパーティション数を1にすることでまとめて出力可能
df = df.coalesce(1)

#テスト用
#print(df.show(3))

# dynamic frame へと戻す
applymapping2 = DynamicFrame.fromDF(df, glueContext, "applymapping2")

resolvechoice2 = ResolveChoice.apply(frame = applymapping2, choice = "make_struct", transformation_ctx = "resolvechoice2")

dropnullfields3 = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = "dropnullfields3")

# Glue データベースへの登録
sink = glueContext.getSink(connection_type="s3", path="s3://バケット名/audit/",enableUpdateCatalog=True, updateBehavior="UPDATE_IN_DATABASE",partitionKeys=["year", "month","day","hour"])
sink.setFormat("parquet", useGlueParquetWriter=True)
sink.setCatalogInfo(catalogDatabase="syslog-converter", catalogTableName="audit")
sink.writeFrame(dropnullfields3)
job.commit()