Athena+embulk で ETL

# 目的

日毎のバッチで、S3にある ParquetファイルをQueryし、その結果をRedshift（DWH）へ入れたかった。

# 背景

* QueryしたいParquetファイルは Data Lake 上にすべて存在していた。
* 必要なParquetファイルはすべて AWS Glue Datacatalog でテーブル構造が定義されていた。
* Queryした結果をRedshift上にテーブル化し、他のアプリケーションで使いたかった。
* Embulkは本番環境はすでにバリバリ使っていた。

# ということで

これだけ揃ってんなら、AthenaでQueryして、結果をEmbulkでRedshiftに入れればいいんでない。

# 作る

## AthenaでQueryして、結果のCSVファイルを確認する

* AWS CLI でAthenaにQueryを投げる。
* Queryを投げるコマンドはQuery要求を出すだけですぐ終了するので終了を待つ。
* jqコマンドを予め yum か何かでインストールしておく

```sh
#!/bin/bash

export RESULT_BACKET="result-bucket"
export RESULT_PATH="result-files"

# 結果待ち
wait_query(){
  max_retry=100
  check_interval=10

  echo -n "sleep ${check_interval} seconds"
  for((i = 0; i <= ${max_retry}; i++)); do
    sleep ${check_interval}

    resultJson=$(aws --profile athena_query athena get-query-execution --query-execution-id $1)
    runningStatus=$(echo ${resultJson} | jq -r '.QueryExecution.Status.State')
    echo -n "."
    if [ "${runningStatus}" == "SUCCEEDED" ]; then
      break
    fi

    if [ "${runningStatus}" == "FAILED" -o "${runningStatus}" == "CANCELLED" ]; then
      echo "query error !!"
      echo "${resultJson}"
      exit -1
    fi
  done
}

### 結果の確認
check_result(){
  json=$(aws --profile athena_query athena get-query-execution --query-execution-id $1 )
  status=$(echo ${json} | jq -r '.QueryExecution.Status.State')

  if [ "$status" == "SUCCEEDED" ]; then
    outputLocation=$(echo ${json} | jq -r '.QueryExecution.ResultConfiguration.OutputLocation')
    echo "query finished!! to ${outputLocation}"
  else
    echo "query result is not found !!"
    echo "${json}"
    exit -1
  fi
}


########## START
### Queryを実行する QueryId を取得する
export QUERY_ID=$(aws --profile athena_query athena start-query-execution \
    --query-string \
      "
  SELECT *
  FROM
    test_schema.table_a
  FULL JOIN
    test_schema.table_b on test_schema.table_a.id = test_schema.table_b = id
      " \
    --result-configuration OutputLocation="s3://${RESULT_BACKET}/${RESULT_PATH}" \
    | jq -r '.QueryExecutionId')

### Query 実行待ち
wait_until_end ${QUERY_ID}

### resulr　file を取得
check_result ${QUERY_ID}
```

### ResultのCSVファイルをEmbulkでRedshiftへInsertする

* source となるCSVファイルのファイル名は ${QUERY_ID}.csv
* embulk の定義ファイルはこちら

```yml:insert.yml

in:
  type: s3
  access_key: {{env.ACCESS_KEY}}
  secret_key: {{env.SECRET_KEY}}
  default_timezone: Asia/Tokyo
  bucket: {{env.RESULT_BACKET}}
  path_prefix: {{env.RESULT_PATH}}/{{env.QUERY_ID}}
  parser:
    type: csv
    delimiter: ","
    skip_header_lines: 1
    columns:
      - {name: id, type: long }
      - {name: point, type: long }
      - {name: count, type: long }
      - {name: regist_date, type: timestamp, format: '%Y-%m-%d %H:%M:%S.%N' }  

out:
  type: redshift
  host: {{ env.REDSHIFT_HOST }}
  port: 5439
  user: {{ env.REDSHIFT_USER }}
  password: {{ env.REDSHIFT_PASSWORD }}
  database: {{ env.REDSHIFT_DATABASE }}
  table: redshift_test_table
  mode: truncate_insert
  aws_auth_method: profile
  aws_profile_name: embulk
  iam_user_name: IAM-Embulk-S3-Accsess
  s3_bucket: embulk-s3-to-redshift
  s3_key_prefix: embulk
  delete_s3_temp_file: false
  default_timezone: "Japan"
  column_options:
    id: {type: 'bigint'}
    point: {type: 'int'}
    count: {type: 'int'}
    regist_date: {type: 'timestamp'}
```

* embulk 実行コマンドライン
* AthenaからQueryしたshに追加する

```sh
/home/embulk/.embulk/bin/embulk run ./insert.yml
```

## 注意点

* AWSの都合でAthenaへのQueryの実行が待たされる場合があるので、日毎のバッチ時実行したいときは実行時間に注意

## その他

* jqコマンド超便利。AWS Cli のresult値取得に超便利。