# 为 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) 部分中的准则，在 Athena 控制台查询编辑器中输入类似以下内容的 DDL 语句。此示例语句创建一个表，其中包含 Amazon VPC 流日志版本 2 到 5 的列，如[流日志记录](https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs.html#flow-log-records)中所示。如果使用不同的列集或列顺序，请相应地修改语句。

   ```
   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. 修改 `LOCATION 's3://amzn-s3-demo-bucket/prefix/AWSLogs/{account_id}/vpcflowlogs/{region_code}/'` 以指向包含您的日志数据的 Amazon S3 存储桶。

1. 在 Athena 控制台中运行查询。查询完成后，Athena 将注册 `vpc_flow_logs` 表，使其中的数据可以供您发出查询。

1. 创建分区以便能够读取数据，如以下示例查询中所示。此示例查询创建指定日期的单个分区。根据需要替换日期和位置的占位符。
**注意**  
此查询仅为您指定的日期创建单个分区。若要自动执行此过程，请使用运行此查询并以这种方式为 `year/month/day` 创建分区的脚本，或使用 `CREATE TABLE` 语句指定[分区投影](vpc-flow-logs-partition-projection.md)。

   ```
   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 地址对它们进行分组，并返回上一周的前 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;
```