AWS Batchで前処理をしてGlue CrawlerでAthenaのスキーマを作成する

1FROM python:3.6.5-alpine
2
3MAINTAINER soudegesu
4
5COPY ./etl /opt/etl
6COPY ./requirements.txt /opt/requirements.txt
7
8RUN pip install --upgrade pip
9RUN pip install -r /opt/requirements.txt

 1resource "aws_batch_compute_environment" "etl" {
 2    compute_environment_name = "etl"
 3    compute_resources {
 4        instance_role = "${ECSインスタンスのRole}"
 5        instance_type = [
 6            ${EC2インスタンスタイプ}"
 7        ]
 8        max_vcpus = 16
 9        min_vcpus = 2
10        desired_vcpus = 2
11        security_group_ids = ["${SecurityGroupのID}"]
12        subnets = ["${SubnetのID}"]
13        type = "EC2"
14    }
15    service_role = "arn:aws:iam::${アカウント番号}:role/service-role/AWSBatchServiceRole"
16    type = "MANAGED"
17}

1resource "aws_batch_job_queue" "etl" {
2    name = "etl"
3    state = "ENABLED"
4    priority = 1
5    compute_environments = [
6        "${aws_batch_compute_environment.etl.arn}"
7    ]
8}

 1resource "aws_batch_job_definition" "etl" {
 2    name = "etl"
 3    type = "container"
 4    container_properties = <<EOF
 5{
 6    "command": ["python",
 7        "/opt/etl/main.py",
 8        "-b",
 9        "Ref::bucket",
10        "-k",
11        "Ref::objKey"
12    ],
13    "environment": [
14         {
15            "name": "TMP",
16            "value": "/tmp"
17         },
18         {
19            "name": "UPLOAD_BUCKET",
20            "value": "${バケット名}"
21         }
22    ],
23    "image": "${var.account_id}.dkr.ecr.${var.region}.amazonaws.com/etl:${var.image_tag}",
24    "memory": 1024,
25    "vcpus": 2,
26    "ulimits": [
27      {
28        "hardLimit": 1024,
29        "name": "nofile",
30        "softLimit": 1024
31      }
32    ]
33}
34EOF
35
36}

 1import json
 2import boto3
 3import logging
 4import os
 5
 6logger = logging.getLogger(__name__)
 7logger.setLevel(logging.INFO)
 8
 9def lambda_handler(event, context):
10
11    job_name = os.environ['JOB_NAME']
12    job_queue = os.environ['JOB_QUEUE']
13    job_definition = os.environ['JOB_DEFINITION']
14    bucket_name = os.environ['BUCKET_NAME']
15    obj_key = os.environ['OBJECT_KEY']
16    parameters = {
17        'bucket': str(bucket_name),
18        'objKey': str(obj_key)
19    }
20
21    batch = boto3.client('batch')
22    try:
23        response = batch.submit_job(jobQueue=job_queue, jobName=job_name, jobDefinition=job_definition, parameters=parameters)
24        jobId = response['jobId']
25        return {
26            'jobId': jobId
27        }
28    except Exception as e:
29        logger.error(e)
30        raise Exception('Error submitting Batch Job')

1    "command": ["python",
2        "/opt/etl/main.py",
3        "-b",
4        "Ref::bucket",
5        "-k",
6        "Ref::objKey"
7    ],

1resource "aws_glue_catalog_database" "sample" {
2    name = "${データベース名}"
3}

 1resource "aws_glue_crawler" "sample" {
 2    database_name = "${aws_glue_catalog_database.sample.name}"
 3    name = "${Crawler名}"
 4    role = "${Glue用のRole}"
 5
 6    s3_target {
 7        path = "${S3バケットのパス}"
 8    }
 9
10    schedule = "cron(0 21 * * ? *)"
11
12    schema_change_policy = {
13        delete_behavior = "DELETE_FROM_DATABASE"
14    }
15}