Import ELB log into Redshift using Data Pipeline

elblog2redshift.json

{
  "objects": [
    {
      "id": "Default",
      "scheduleType": "TIMESERIES",
      "failureAndRerunMode": "CASCADE",
      "name": "Default",
      "pipelineLogUri": "s3://path_to_log",
      "role": "DataPipelineDefaultRole",
      "resourceRole": "DataPipelineDefaultResourceRole"
    },
    {
      "id": "ScheduleId1",
      "name": "DefaultSchedule1",
      "startAt": "FIRST_ACTIVATION_DATE_TIME",
      "type": "Schedule",
      "period": "15 Minutes"
    },
    {
      "id": "S3DataNodeElbLog",
      "name": "S3DataNodeElbLog",
      "type": "S3DataNode",
      "directoryPath": "s3://path_to_elb_log/",
      "schedule": {
        "ref": "ScheduleId1"
      }
    },
    {
      "id": "RedshiftDatabaseId1",
      "name": "DefaultRedshiftDatabase1",
      "type": "RedshiftDatabase",
      "clusterId": "hoge",
      "databaseName": "dev",
      "username": "hogehoge",
      "*password": "fugafuga"
    },
    {
      "id": "RedshiftDataNodeElbLog",
      "name": "RedshiftDataNodeElbLog",
      "type": "RedshiftDataNode",
      "database": {
        "ref": "RedshiftDatabaseId1"
      },
      "tableName": "elb_access_logs",
      "createTableSql": "CREATE TABLE elb_access_logs ( request_timestamp DateTime encode lzo, elb varchar(30) encode lzo, client_port varchar(22) encode lzo, backend_port varchar(22) encode lzo, request_processing_time FLOAT encode bytedict, backend_processing_time FLOAT encode bytedict, response_prosessing_time FLOAT encode bytedict, elb_status_code varchar(3) encode lzo, backend_status_code varchar(3) encode lzo, received_bytes BIGINT encode lzo, sent_bytes BIGINT encode lzo, request varchar(MAX) encode lzo)sortkey(request_timestamp) ;",
      "schedule": {
        "ref": "ScheduleId1"
      }
    },
    {
      "id": "RedshiftCopyActivityElbLog",
      "name": "RedshiftCopyActivityElbLog",
      "type": "RedshiftCopyActivity",
      "input": {
        "ref": "S3DataNodeElbLog"
      },
      "commandOptions": [
        "DELIMITER ' '",
        "TRUNCATECOLUMNS",
        "TRIMBLANKS",
        "REMOVEQUOTES",
        "TIMEFORMAT as 'auto'",
        "ACCEPTINVCHARS",
        "MAXERROR as 100000"
      ],
      "insertMode": "TRUNCATE",
      "runsOn": {
        "ref": "Ec2ResourceId1"
      },
      "schedule": {
        "ref": "ScheduleId1"
      },
      "output": {
        "ref": "RedshiftDataNodeElbLog"
      }
    },
    {
      "id": "Ec2ResourceId1",
      "name": "DefaultEc2Resource1",
      "type": "Ec2Resource",
      "terminateAfter": "1 HOURS",
      "schedule": {
        "ref": "ScheduleId1"
      },
      "securityGroups": "default",
      "logUri": "s3://path_to_log",
      "role": "DataPipelineDefaultRole",
      "resourceRole": "DataPipelineDefaultResourceRole"
    }
  ]
}