# 파티션 프로젝션을 사용하여 Amazon VPC 흐름 로그용 테이블 생성 및 쿼리
<a name="vpc-flow-logs-partition-projection"></a>

다음과 같이 `CREATE TABLE`을 사용하여 테이블을 생성하고, 테이블을 분할하고, [파티션 프로젝션](partition-projection.md)을 사용하여 자동으로 파티션을 채웁니다. 예에서 테이블 이름 `test_table_vpclogs`를 테이블 이름으로 바꿉니다. `LOCATION` 절을 편집하여 Amazon VPC 로그 데이터가 포함된 Amazon S3 버킷을 지정합니다.

다음 `CREATE TABLE` 문은 Hive 스타일이 아닌 분할 형식으로 전달되는 VPC 흐름 로그입니다. 이 예제에서는 다중 계정 집계를 허용합니다. 여러 계정의 VPC 흐름 로그를 하나의 Amazon S3 버킷으로 중앙 집중화하려는 경우 Amazon S3 경로에 계정 ID를 입력해야 합니다.

```
CREATE EXTERNAL TABLE IF NOT EXISTS test_table_vpclogs (
  version int,
  account_id string,
  interface_id string,
  srcaddr string,
  dstaddr string,
  srcport int,
  dstport int,
  protocol bigint,
  packets bigint,
  bytes bigint,
  start bigint,
  `end` bigint,
  action string,
  log_status string,
  vpc_id string,
  subnet_id string,
  instance_id string,
  tcp_flags int,
  type string,
  pkt_srcaddr string,
  pkt_dstaddr string,
  az_id string,
  sublocation_type string,
  sublocation_id string,
  pkt_src_aws_service string,
  pkt_dst_aws_service string,
  flow_direction string,
  traffic_path int
)
PARTITIONED BY (accid string, region string, day string)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ' '
LOCATION '$LOCATION_OF_LOGS'
TBLPROPERTIES
(
"skip.header.line.count"="1",
"projection.enabled" = "true",
"projection.accid.type" = "enum",
"projection.accid.values" = "$ACCID_1,$ACCID_2",
"projection.region.type" = "enum",
"projection.region.values" = "$REGION_1,$REGION_2,$REGION_3",
"projection.day.type" = "date",
"projection.day.range" = "$START_RANGE,NOW",
"projection.day.format" = "yyyy/MM/dd",
"storage.location.template" = "s3://amzn-s3-demo-bucket/AWSLogs/${accid}/vpcflowlogs/${region}/${day}"
)
```

## test\$1table\$1vpclogs에 대한 쿼리 예
<a name="query-examples-vpc-logs-pp"></a>

다음 예제 쿼리는 선행하는 `CREATE TABLE` 문에 의해 생성된 `test_table_vpclogs`를 쿼리합니다. 쿼리에서 `test_table_vpclogs`를 자체 테이블 이름으로 바꿉니다. 사용자의 요구 사항에 따라 열 값과 기타 변수를 수정합니다.

지정된 기간 동안 처음 100개의 액세스 로그 항목을 시간순으로 반환하려면 다음과 같은 쿼리를 실행합니다.

```
SELECT *
FROM test_table_vpclogs
WHERE day >= '2021/02/01' AND day < '2021/02/28'
ORDER BY day ASC
LIMIT 100
```

지정된 기간 동안 상위 10개 HTTP 패킷을 수신하는 서버를 보려면 다음과 같은 쿼리를 실행합니다. 이 쿼리는 지난 주부터 HTTPS 포트 443에서 수신한 패킷 수를 계산하고, 대상 IP 주소별로 그룹화한 다음, 상위 10개를 반환합니다.

```
SELECT SUM(packets) AS packetcount, 
       dstaddr
FROM test_table_vpclogs
WHERE dstport = 443
  AND day >= '2021/03/01'
  AND day < '2021/03/31'
GROUP BY dstaddr
ORDER BY packetcount DESC
LIMIT 10
```

지정된 기간 동안 생성된 로그를 반환하려면 다음과 같은 쿼리를 실행합니다.

```
SELECT interface_id,
       srcaddr,
       action,
       protocol,
       to_iso8601(from_unixtime(start)) AS start_time,
       to_iso8601(from_unixtime("end")) AS end_time
FROM test_table_vpclogs
WHERE DAY >= '2021/04/01'
  AND DAY < '2021/04/30'
```

지정된 기간 사이에 소스 IP 주소에 대한 액세스 로그를 반환하려면 다음과 같은 쿼리를 실행합니다.

```
SELECT *
FROM test_table_vpclogs
WHERE srcaddr = '10.117.1.22'
  AND day >= '2021/02/01'
  AND day < '2021/02/28'
```

거부된 TCP 연결을 나열하려면 다음과 같은 쿼리를 실행합니다.

```
SELECT day,
       interface_id,
       srcaddr,
       action,
       protocol
FROM test_table_vpclogs
WHERE action = 'REJECT' AND protocol = 6 AND day >= '2021/02/01' AND day < '2021/02/28'
LIMIT 10
```

`10.117`로 시작하는 IP 주소 범위에 대한 액세스 로그를 반환하려면 다음과 같은 쿼리를 실행합니다.

```
SELECT *
FROM test_table_vpclogs
WHERE split_part(srcaddr,'.', 1)='10'
  AND split_part(srcaddr,'.', 2) ='117'
```

지정된 기간 사이에 대상 IP 주소에 대한 액세스 로그를 반환하려면 다음과 같은 쿼리를 실행합니다.

```
SELECT *
FROM test_table_vpclogs
WHERE dstaddr = '10.0.1.14'
  AND day >= '2021/01/01'
  AND day < '2021/01/31'
```