diff --git a/aws/resource_aws_flow_log.go b/aws/resource_aws_flow_log.go
index 082e90db453..5ed03640695 100644
--- a/aws/resource_aws_flow_log.go
+++ b/aws/resource_aws_flow_log.go
@@ -101,6 +101,15 @@ func resourceAwsFlowLog() *schema.Resource {
 				ForceNew: true,
 				Computed: true,
 			},
+
+			"max_aggregation_interval": {
+				Type:         schema.TypeInt,
+				Optional:     true,
+				ForceNew:     true,
+				Default:      600,
+				ValidateFunc: validation.IntInSlice([]int{60, 600}),
+			},
+
 			"tags": tagsSchema(),
 		},
 	}
@@ -150,10 +159,15 @@ func resourceAwsLogFlowCreate(d *schema.ResourceData, meta interface{}) error {
 	if v, ok := d.GetOk("log_group_name"); ok && v != "" {
 		opts.LogGroupName = aws.String(v.(string))
 	}
+
 	if v, ok := d.GetOk("log_format"); ok && v != "" {
 		opts.LogFormat = aws.String(v.(string))
 	}
 
+	if v, ok := d.GetOk("max_aggregation_interval"); ok {
+		opts.MaxAggregationInterval = aws.Int64(int64(v.(int)))
+	}
+
 	if v, ok := d.GetOk("tags"); ok && len(v.(map[string]interface{})) > 0 {
 		opts.TagSpecifications = ec2TagSpecificationsFromMap(d.Get("tags").(map[string]interface{}), ec2.ResourceTypeVpcFlowLog)
 	}
@@ -205,6 +219,7 @@ func resourceAwsLogFlowRead(d *schema.ResourceData, meta interface{}) error {
 	d.Set("log_group_name", fl.LogGroupName)
 	d.Set("iam_role_arn", fl.DeliverLogsPermissionArn)
 	d.Set("log_format", fl.LogFormat)
+	d.Set("max_aggregation_interval", fl.MaxAggregationInterval)
 	var resourceKey string
 	if strings.HasPrefix(*fl.ResourceId, "vpc-") {
 		resourceKey = "vpc_id"
diff --git a/aws/resource_aws_flow_log_test.go b/aws/resource_aws_flow_log_test.go
index 67207403aee..44c09455775 100644
--- a/aws/resource_aws_flow_log_test.go
+++ b/aws/resource_aws_flow_log_test.go
@@ -34,6 +34,7 @@ func TestAccAWSFlowLog_VPCID(t *testing.T) {
 					resource.TestCheckResourceAttr(resourceName, "log_destination", ""),
 					resource.TestCheckResourceAttr(resourceName, "log_destination_type", "cloud-watch-logs"),
 					resource.TestCheckResourceAttrPair(resourceName, "log_group_name", cloudwatchLogGroupResourceName, "name"),
+					resource.TestCheckResourceAttr(resourceName, "max_aggregation_interval", "600"),
 					resource.TestCheckResourceAttr(resourceName, "traffic_type", "ALL"),
 					resource.TestCheckResourceAttrPair(resourceName, "vpc_id", vpcResourceName, "id"),
 				),
@@ -105,6 +106,7 @@ func TestAccAWSFlowLog_SubnetID(t *testing.T) {
 					resource.TestCheckResourceAttr(resourceName, "log_destination", ""),
 					resource.TestCheckResourceAttr(resourceName, "log_destination_type", "cloud-watch-logs"),
 					resource.TestCheckResourceAttrPair(resourceName, "log_group_name", cloudwatchLogGroupResourceName, "name"),
+					resource.TestCheckResourceAttr(resourceName, "max_aggregation_interval", "600"),
 					resource.TestCheckResourceAttrPair(resourceName, "subnet_id", subnetResourceName, "id"),
 					resource.TestCheckResourceAttr(resourceName, "traffic_type", "ALL"),
 				),
@@ -195,6 +197,33 @@ func TestAccAWSFlowLog_LogDestinationType_S3_Invalid(t *testing.T) {
 	})
 }
 
+func TestAccAWSFlowLog_LogDestinationType_MaxAggregationInterval(t *testing.T) {
+	var flowLog ec2.FlowLog
+	resourceName := "aws_flow_log.test"
+	rName := acctest.RandomWithPrefix("tf-acc-test")
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckFlowLogDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccFlowLogConfig_MaxAggregationInterval(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckFlowLogExists(resourceName, &flowLog),
+					testAccCheckAWSFlowLogAttributes(&flowLog),
+					resource.TestCheckResourceAttr(resourceName, "max_aggregation_interval", "60"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
 func TestAccAWSFlowLog_tags(t *testing.T) {
 	var flowLog ec2.FlowLog
 	resourceName := "aws_flow_log.test"
@@ -491,6 +520,7 @@ resource "aws_flow_log" "test" {
 }
 `, rName)
 }
+
 func testAccFlowLogConfig_LogFormat(rName string) string {
 	return testAccFlowLogConfigBase(rName) + fmt.Sprintf(`
 resource "aws_iam_role" "test" {
@@ -523,7 +553,7 @@ resource "aws_s3_bucket" "test" {
 	bucket        = %[1]q
 	force_destroy = true
   }
-  
+
 
 resource "aws_flow_log" "test" {
   log_destination      = "${aws_s3_bucket.test.arn}"
@@ -621,3 +651,43 @@ resource "aws_flow_log" "test" {
 }
 `, rName, tagKey1, tagValue1, tagKey2, tagValue2)
 }
+
+func testAccFlowLogConfig_MaxAggregationInterval(rName string) string {
+	return testAccFlowLogConfigBase(rName) + fmt.Sprintf(`
+resource "aws_iam_role" "test" {
+  name = %[1]q
+
+  assume_role_policy = <<EOF
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Principal": {
+        "Service": [
+          "ec2.amazonaws.com"
+        ]
+      },
+      "Action": [
+        "sts:AssumeRole"
+      ]
+    }
+  ]
+}
+EOF
+}
+
+resource "aws_cloudwatch_log_group" "test" {
+  name = %[1]q
+}
+
+resource "aws_flow_log" "test" {
+  iam_role_arn   = "${aws_iam_role.test.arn}"
+  log_group_name = "${aws_cloudwatch_log_group.test.name}"
+  traffic_type   = "ALL"
+  vpc_id         = "${aws_vpc.test.id}"
+
+  max_aggregation_interval = 60
+}
+`, rName)
+}
diff --git a/website/docs/r/flow_log.html.markdown b/website/docs/r/flow_log.html.markdown
index 3524f6c479c..8b0cd77dbe4 100644
--- a/website/docs/r/flow_log.html.markdown
+++ b/website/docs/r/flow_log.html.markdown
@@ -102,6 +102,10 @@ The following arguments are supported:
 * `subnet_id` - (Optional) Subnet ID to attach to
 * `vpc_id` - (Optional) VPC ID to attach to
 * `log_format` - (Optional) The fields to include in the flow log record, in the order in which they should appear.
+* `max_aggregation_interval` - (Optional) The maximum interval of time
+  during which a flow of packets is captured and aggregated into a flow
+  log record. Valid Values: `60` seconds (1 minute) or `600` seconds (10
+  minutes). Default: `600`.
 * `tags` - (Optional) Key-value mapping of resource tags
 
 ## Attributes Reference