Athena を使用して Amazon VPC フローログを分析する方法を教えてください。

解決策

Athena コンソールのクエリエディタを使用すると、データベースを作成し、VPC フローログ用のテーブルを作成し、サンプルクエリを実行できます。次に、フローログを使用してネットワークトラフィックのパターンを分析し、Amazon VPC ネットワーク全体の脅威とリスクを特定します。

Athena コンソールのクエリエディタでデータベースを作成する

Athena コンソールのクエリエディタを開き、次のコマンドを実行します。

CREATE DATABASE test_db_vpclogs;

注: test_db_vpclogs を、データベースに使用する名前に置き換えてください。

重要: フローログが保存されている Amazon Simple Storage Service (Amazon S3) バケットと同じ AWS リージョンにデータベースを作成することをお勧めします。

データベース内のフローログ用のテーブルを作成する

Athena コンソールを開きます。次に、ナビゲーションペインで、[クエリエディタ] をクリックします。Athena コンソールのクエリエディタで、次の例のようなコマンドを実行します。

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 (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}"
)

注: **test_table_vpclogs ** をテーブルに使用する名前に置き換えてください。

LOCATION パラメータを変更して、ログデータを含む Amazon S3 バケットを指すようにします。projection.region.values を VPC フローログがある AWS リージョンに更新します。projection.date.range の値を、データで利用可能な最も古いログに設定します。例えば、ログが 2021 年 1 月 1 日に開始される場合は、projection.date.range を '2021/01/01,NOW' に設定します。正確なデータ予測を行うには、ログの実際の開始日に応じて値を変更します。上記のコマンドは、パーティションプロジェクションを使用してテーブルを作成し、テーブルをパーティション化し、自動的に各パーティションに入力します。投影されたパーティションが Amazon S3 に存在しない場合も、Athena はそのパーティションを投影します。クエリではパーティション化した属性を使用することをお勧めします。

フローログ用のテーブルに SQL ステートメントを実行する

Athena コンソールを開きます。次に、ナビゲーションペインで、[クエリエディタ] をクリックします。Athena コンソールのクエリエディタを使用して、テーブルに対する SQL ステートメントを実行します。クエリを保存したり、以前のクエリを確認したり、クエリ結果を .csv 形式でダウンロードしたりすることができます。

クエリ例

注: 次のクエリ例では、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;

指定した期間に最多の HTTPS パケットを受信した 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'

関連情報

フローログを使用して VPC のトラフィックをモニタリングするにはどうすればよいですか?

Analyzing VPC Flow Logs using Amazon Athena and Amazon QuickSight (Amazon Athena と Amazon QuickSight を使用した VPC フローログの分析)