# Amazon VPC フローログをクエリする
<a name="vpc-flow-logs"></a>

Amazon Virtual Private Cloud フローログは、VPC 内のネットワークインターフェイス間で送受信される IP トラフィックに関する情報をキャプチャします。このログを使用してネットワークトラフィックのパターンを調査し、VPC ネットワーク全体の脅威やリスクを特定します。

Amazon VPC フローログのクエリを実行するには、2 つの方法があります。

****
+ **Amazon VPC コンソール** — Amazon VPC コンソールの Athena 統合機能を使用して、パーティションを含む Athena データベース、ワークグループ、およびフローログテーブルを作成する CloudFormation テンプレートを生成します。テンプレートでは、[事前定義されたフローログのクエリ](https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs-athena.html#predefined-queries)のセットも作成します。これは、VPC を通過するトラフィックに関するインサイトの取得に使用できます。

  このアプローチの詳細については、「*Amazon VPC ユーザーガイド*」の「[Amazon Athena を使用したフローログのクエリ](https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs-athena.html)」を参照してください。
+ **Amazon Athena コンソール** — Athena コンソールでテーブルやクエリを直接作成します。詳細については、引き続きこのページを参照してください。

Athena でログのクエリを開始する前に、[VPC フローログを有効化](https://docs.aws.amazon.com/AmazonVPC/latest/UserGuide/flow-logs.html)し、それらが Amazon S3 バケットに保存されるように設定します。ログを作成したら、それを数分間実行していくらかのデータを収集します。ログは、Athena で直接クエリできる GZIP 圧縮形式で作成されます。

VPC フローログの作成時に、フローログで返すフィールドおよびフィールドが表示される順序を指定する際は、カスタム形式を使用できます。フローログレコード詳細については、「*Amazon VPC ユーザーガイド*」の「[フローログレコード](https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs.html#flow-log-records)」を参照してください。

## 考慮事項と制限事項
<a name="vpc-flow-logs-common-considerations"></a>

Athena で Amazon VPC フローログのテーブルを作成する際は、次の点に注意してください。
+ デフォルトでは、Parquet は名前で列にアクセスします。詳細については、「[スキーマの更新を処理する](handling-schema-updates-chapter.md)」を参照してください。
+ Athena の列名には、フローログレコードの名前を使用します。Athena スキーマの列名は、Amazon VPC フローログのフィールド名と完全に一致する必要があります。ただし、これらには次の違いがあります。
  + Amazon VPC のログフィールド名のハイフンは、Athena の列名ではアンダースコアに置き換えられます。Athena のデータベース名、テーブル名、および列名で使用できる文字については、「[データベース、テーブル、列に名前を付ける](tables-databases-columns-names.md)」を参照してください。
  + バックティックで囲むことで、Athena で[予約されたキーワード](reserved-words.md)であるフローログのレコード名をエスケープします。
+ VPC フローログは AWS アカウント 固有です。ログファイルを Amazon S3 に発行すると、Amazon VPC が Amazon S3 で作成するパスには、フローログの作成に使用された AWS アカウント の ID が含まれます。詳細については、「*Amazon VPC ユーザーガイド*」の「[フローログを Amazon S3 に発行する](https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs-s3.html)」を参照してください。

**Topics**
+ [考慮事項と制限事項](#vpc-flow-logs-common-considerations)
+ [Amazon VPC フローログのテーブルを作成し、クエリする](vpc-flow-logs-create-table-statement.md)
+ [Apache Parquet 形式でフローログのテーブルを作成する](vpc-flow-logs-parquet.md)
+ [パーティション射影を使用して Amazon VPC フローログのテーブルを作成およびクエリする](vpc-flow-logs-partition-projection.md)
+ [パーティション射影を使用して Apache Parquet 形式でフローログのテーブルを作成する](vpc-flow-logs-partition-projection-parquet-example.md)
+ [その他のリソース](query-examples-vpc-logs-additional-resources.md)

# Amazon VPC フローログのテーブルを作成し、クエリする
<a name="vpc-flow-logs-create-table-statement"></a>

次の手順では、Amazon VPC フローログ用の Amazon VPC テーブルを作成します。カスタム形式でフローログを作成する場合は、フローログの作成時に指定したフィールドと一致するフィールドを、それらに指定したものと同じ順序で使用してテーブルを作成します。

**Amazon VPC フローログ用の Athena テーブルを作成するには**

1. [考慮事項と制限事項](vpc-flow-logs.md#vpc-flow-logs-common-considerations) セクションのガイドラインに従って、次のような DDL ステートメントを Athena コンソールクエリエディタに入力します。サンプルステートメントは、「[フローログレコード](https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs.html#flow-log-records)」に記載されているように、Amazon VPC フローログのバージョン 2 から 5 の列を持つテーブルを作成します。異なる列のセットまたは列の順序を使用する場合は、必要に応じてステートメントを変更します。

   ```
   CREATE EXTERNAL TABLE IF NOT EXISTS `vpc_flow_logs` (
     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,
     region 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 (`date` date)
   ROW FORMAT DELIMITED
   FIELDS TERMINATED BY ' '
   LOCATION 's3://amzn-s3-demo-bucket/prefix/AWSLogs/{account_id}/vpcflowlogs/{region_code}/'
   TBLPROPERTIES ("skip.header.line.count"="1");
   ```

   次の点に注意してください。
   + このクエリは `ROW FORMAT DELIMITED` を指定し、SerDe の指定が省略されます。これは、クエリが [CSV、TSV、カスタム区切りファイル用の Lazy Simple SerDe](lazy-simple-serde.md) を使用しているということです。このクエリでは、フィールドはスペースで終了します。
   + `PARTITIONED BY` 句は、`date` 型を使用します。これにより、クエリで数学演算子を使用して、特定の日付より古いものまたは新しいものを選択できます。
**注記**  
`date` は DDL ステートメントの予約済みキーワードであるため、バックティック文字でエスケープされます。詳細については、「[クエリで予約キーワードをエスケープする](reserved-words.md)」を参照してください。
   + 異なるカスタム形式の VPC フローログの場合、フローログの作成時に指定したフィールドと一致するようにフィールドを変更します。

1. ログデータが含まれる Amazon S3 バケットをポイントするように `LOCATION 's3://amzn-s3-demo-bucket/prefix/AWSLogs/{account_id}/vpcflowlogs/{region_code}/'` を変更します。

1. Athena コンソールでクエリを実行します。クエリが完了すると、Athena が `vpc_flow_logs` テーブルを登録し、その中のデータに対してクエリを発行できるように準備します。

1. 次のサンプルクエリのように、パーティションを作成してデータを読み取れるようにします。このクエリは、指定日の 1 つのパーティションを作成します。必要に応じて、日付と場所のプレースホルダーを置き換えます。
**注記**  
このクエリは、指定日に対して単一のパーティションのみを作成します。プロセスを自動化するには、このクエリを実行し、この方法で `year/month/day` にパーティションを作成するスクリプトを使用するか、[パーティション射影](vpc-flow-logs-partition-projection.md)を指定する `CREATE TABLE` ステートメントを使用します。

   ```
   ALTER TABLE vpc_flow_logs
   ADD PARTITION (`date`='YYYY-MM-dd')
   LOCATION 's3://amzn-s3-demo-bucket/prefix/AWSLogs/{account_id}/vpcflowlogs/{region_code}/YYYY/MM/dd';
   ```

## vpc\$1flow\$1logs テーブルのクエリ例
<a name="query-examples-vpc-logs"></a>

Athena コンソールのクエリエディタを使用して、作成したテーブルで SQL ステートメントを実行します。クエリを保存したり、以前のクエリを表示したり、クエリ結果を CSV 形式でダウンロードできます。次の例では、`vpc_flow_logs` をテーブルの名前に置き換えます。また、要件に応じて列の値やその他の変数を変更します。

次のクエリ例では、指定された日付に対して最大 100 のフローログを一覧表示します。

```
SELECT * 
FROM vpc_flow_logs 
WHERE date = DATE('2020-05-04') 
LIMIT 100;
```

次のクエリは、すべての拒否された TCP 接続を一覧表示します。新しく作成した日付パーティション列 `date` を使用して、該当イベントが発生した週の曜日を抽出します。

```
SELECT day_of_week(date) AS
  day,
  date,
  interface_id,
  srcaddr,
  action,
  protocol
FROM vpc_flow_logs
WHERE action = 'REJECT' AND protocol = 6
LIMIT 100;
```

最大数の HTTPS リクエストを受信しているサーバーを確認するには、次のクエリを使用します。HTTPS ポート 443 で受信したパケット数をカウントし、送信先 IP アドレス別にグループ分けして、過去 1 週間の上位 10 のサーバーを返します。

```
SELECT SUM(packets) AS
  packetcount,
  dstaddr
FROM vpc_flow_logs
WHERE dstport = 443 AND date > current_date - interval '7' day
GROUP BY dstaddr
ORDER BY packetcount DESC
LIMIT 10;
```

# Apache Parquet 形式でフローログのテーブルを作成する
<a name="vpc-flow-logs-parquet"></a>

次の手順では、Apache Parquet 形式で Amazon VPC フローログ用の Amazon VPC テーブルを作成します。

**Parquet 形式で Amazon VPC フローログ用の Athena テーブルを作成するには**

1. [考慮事項と制限事項](vpc-flow-logs.md#vpc-flow-logs-common-considerations) セクションのガイドラインに従って、次のような DDL ステートメントを Athena コンソールクエリエディタに入力します。サンプルステートメントは、「[フローログレコード](https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs.html#flow-log-records)」に記載されているように、1 時間ごとに Hive パーティションされる Parquet 形式で Amazon VPC フローログのバージョン 2 から 5 の列を持つテーブルを作成します。1 時間ごとのパーティションがない場合は、`PARTITIONED BY` 句から `hour` を削除します。

   ```
   CREATE EXTERNAL TABLE IF NOT EXISTS vpc_flow_logs_parquet (
     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,
     region 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 (
     `aws-account-id` string,
     `aws-service` string,
     `aws-region` string,
     `year` string, 
     `month` string, 
     `day` string,
     `hour` string
   )
   ROW FORMAT SERDE 
     'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
   STORED AS INPUTFORMAT 
     'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' 
   OUTPUTFORMAT 
     'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
   LOCATION
     's3://amzn-s3-demo-bucket/prefix/AWSLogs/'
   TBLPROPERTIES (
     'EXTERNAL'='true', 
     'skip.header.line.count'='1'
     )
   ```

1. ログデータが含まれる Amazon S3 パスをポイントするようにサンプル `LOCATION 's3://amzn-s3-demo-bucket/prefix/AWSLogs/'` を変更します。

1. Athena コンソールでクエリを実行します。

1. Hive 互換形式のデータの場合、Athena コンソールで次のコマンドを実行して、メタストアの Hive パーティションを更新およびロードします。クエリが完了すると、`vpc_flow_logs_parquet` テーブルでデータをクエリできます。

   ```
   MSCK REPAIR TABLE vpc_flow_logs_parquet
   ```

   Hive 互換データを使用していない場合は、[ALTER TABLE ADD PARTITION](alter-table-add-partition.md) を実行してパーティションをロードします。

Athena を使用して、Parquet 形式の Amazon VPC フローログをクエリする方法については、「*AWS ビッグデータブログ*」の「[Apache Parquet 形式の VPC フローログを使用して、ネットワーク分析のパフォーマンスを最適化し、そのコストを削減する](https://aws.amazon.com/blogs/big-data/optimize-performance-and-reduce-costs-for-network-analytics-with-vpc-flow-logs-in-apache-parquet-format/)」を参照してください。

# パーティション射影を使用して 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'
```

# パーティション射影を使用して Apache Parquet 形式でフローログのテーブルを作成する
<a name="vpc-flow-logs-partition-projection-parquet-example"></a>

次の VPC フローログのパーティション射影 `CREATE TABLE` ステートメントは Apache Parquet 形式であり、Hive と互換性がなく、曜日ではなく時間と日付でパーティショニングされています。サンプルのテーブル名 `test_table_vpclogs_parquet` をそのテーブルの名前に置き換えます。Amazon VPC ログデータが含まれている Amazon S3 バケットを指定するように `LOCATION` 句を編集します。

```
CREATE EXTERNAL TABLE IF NOT EXISTS test_table_vpclogs_parquet (
  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 (region string, date string, hour string)
ROW FORMAT SERDE
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION 's3://amzn-s3-demo-bucket/prefix/AWSLogs/{account_id}/vpcflowlogs/'
TBLPROPERTIES (
"EXTERNAL"="true",
"skip.header.line.count" = "1",
"projection.enabled" = "true",
"projection.region.type" = "enum",
"projection.region.values" = "us-east-1,us-west-2,ap-south-1,eu-west-1",
"projection.date.type" = "date",
"projection.date.range" = "2021/01/01,NOW",
"projection.date.format" = "yyyy/MM/dd",
"projection.hour.type" = "integer",
"projection.hour.range" = "00,23",
"projection.hour.digits" = "2",
"storage.location.template" = "s3://amzn-s3-demo-bucket/prefix/AWSLogs/${account_id}/vpcflowlogs/${region}/${date}/${hour}"
)
```

# その他のリソース
<a name="query-examples-vpc-logs-additional-resources"></a>

Athena を使用して VPC フローログを分析する方法の詳細については、次の「AWS Big Data Blog」の記事を参照してください。
+ [ポイントアンドクリック Amazon Athena インテグレーションを使用した VPC フローログの分析](https://aws.amazon.com/blogs/networking-and-content-delivery/analyze-vpc-flow-logs-with-point-and-click-amazon-athena-integration/) 
+ [Amazon Athena および Quick を使用した VPC フローログの分析](https://aws.amazon.com/blogs/big-data/analyzing-vpc-flow-logs-using-amazon-athena-and-amazon-quicksight/)
+ [Apache Parquet 形式の VPC フローログでパフォーマンスを最適化し、ネットワーク分析のコストを削減する](https://aws.amazon.com/blogs/big-data/optimize-performance-and-reduce-costs-for-network-analytics-with-vpc-flow-logs-in-apache-parquet-format/)