# パーティション射影を使用して Amazon VPC フローログのテーブルを作成およびクエリする
<a name="vpc-flow-logs-partition-projection"></a>

次のような `CREATE TABLE` ステートメントを使用してテーブルを作成およびパーティションし、[パーティション射影](partition-projection.md)を使用してパーティションを自動的に入力します。サンプルのテーブル名 `test_table_vpclogs` をそのテーブルの名前に置き換えます。Amazon VPC ログデータが含まれている Amazon S3 バケットを指定するように `LOCATION` 句を編集します。

次は、非 Hive スタイルのパーティション形式で配信される VPC フローログ用の `CREATE TABLE` ステートメントです。この例では、マルチアカウント集計が可能です。複数のアカウントからの VPC フローログを 1 つの Amazon S3 バケットに集中管理する場合は、アカウント ID を Amazon S3 パスに入力する必要があります。

```
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 アドレス別にグループ分けして、過去 1 週間の上位 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'
```