diff --git a/aws/resource_aws_emr_cluster.go b/aws/resource_aws_emr_cluster.go
index 7f2f8caa1809..fff99c536df7 100644
--- a/aws/resource_aws_emr_cluster.go
+++ b/aws/resource_aws_emr_cluster.go
@@ -1262,6 +1262,7 @@ func flattenInstanceGroups(igs []*emr.InstanceGroup) (*schema.Set, error) {
 }
 
 func flattenEBSConfig(ebsBlockDevices []*emr.EbsBlockDevice) *schema.Set {
+
 	ebsConfig := make([]interface{}, 0)
 	for _, ebs := range ebsBlockDevices {
 		ebsAttrs := make(map[string]interface{})
diff --git a/aws/resource_aws_emr_instance_group.go b/aws/resource_aws_emr_instance_group.go
index 8f0589b51651..2dc53c6bf515 100644
--- a/aws/resource_aws_emr_instance_group.go
+++ b/aws/resource_aws_emr_instance_group.go
@@ -56,26 +56,32 @@ func resourceAwsEMRInstanceGroup() *schema.Resource {
 				ForceNew: true,
 			},
 			"ebs_config": {
-				Type:     schema.TypeSet,
+				Type:     schema.TypeList,
 				Optional: true,
+				Computed: true,
 				ForceNew: true,
+				MinItems: 1,
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"iops": {
 							Type:     schema.TypeInt,
+							ForceNew: true,
 							Optional: true,
 						},
 						"size": {
 							Type:     schema.TypeInt,
+							ForceNew: true,
 							Required: true,
 						},
 						"type": {
 							Type:         schema.TypeString,
 							Required:     true,
+							ForceNew:     true,
 							ValidateFunc: validateAwsEmrEbsVolumeType(),
 						},
 						"volumes_per_instance": {
 							Type:     schema.TypeInt,
+							ForceNew: true,
 							Optional: true,
 						},
 					},
@@ -92,7 +98,6 @@ func resourceAwsEMRInstanceGroup() *schema.Resource {
 				ForceNew: true,
 				Default:  emr.InstanceGroupTypeTask,
 				ValidateFunc: validation.StringInSlice([]string{
-					emr.InstanceGroupTypeCore,
 					emr.InstanceGroupTypeTask}, false),
 			},
 			"instance_type": {
@@ -138,8 +143,10 @@ func resourceAwsEMRInstanceGroupCreate(d *schema.ResourceData, meta interface{})
 		groupConfig.AutoScalingPolicy = autoScalingPolicy
 	}
 
+	groupConfig.Market = aws.String(emr.MarketTypeOnDemand)
 	if v, ok := d.GetOk("bid_price"); ok {
 		groupConfig.BidPrice = aws.String(v.(string))
+		groupConfig.Market = aws.String(emr.MarketTypeSpot)
 	}
 
 	params := &emr.AddInstanceGroupsInput{
@@ -209,6 +216,7 @@ func resourceAwsEMRInstanceGroupRead(d *schema.ResourceData, meta interface{}) e
 	d.Set("autoscaling_policy", autoscalingPolicyString)
 
 	d.Set("bid_price", ig.BidPrice)
+	d.Set("ebs_config", flattenEBSConfig(ig.EbsBlockDevices))
 	d.Set("instance_count", ig.RequestedInstanceCount)
 	d.Set("instance_role", ig.InstanceGroupType)
 	d.Set("instance_type", ig.InstanceType)
@@ -238,6 +246,8 @@ func resourceAwsEMRInstanceGroupUpdate(d *schema.ResourceData, meta interface{})
 			},
 		}
 
+		// Give me some EBS config stuff.
+
 		_, err := conn.ModifyInstanceGroups(params)
 		if err != nil {
 			return fmt.Errorf("error modifying EMR Instance Group (%s): %s", d.Id(), err)
diff --git a/aws/resource_aws_emr_instance_group_test.go b/aws/resource_aws_emr_instance_group_test.go
index fbd91bf3eafc..d719f2b533a6 100644
--- a/aws/resource_aws_emr_instance_group_test.go
+++ b/aws/resource_aws_emr_instance_group_test.go
@@ -16,6 +16,7 @@ import (
 func TestAccAWSEMRInstanceGroup_basic(t *testing.T) {
 	var ig emr.InstanceGroup
 	rInt := acctest.RandInt()
+
 	resourceName := "aws_emr_instance_group.task"
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -26,7 +27,10 @@ func TestAccAWSEMRInstanceGroup_basic(t *testing.T) {
 				Config: testAccAWSEmrInstanceGroupConfig_basic(rInt),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckAWSEmrInstanceGroupExists(resourceName, &ig),
-					resource.TestCheckResourceAttr("aws_emr_instance_group.task", "instance_role", emr.InstanceGroupTypeTask),
+					resource.TestCheckResourceAttr(resourceName, "autoscaling_policy", ""),
+					resource.TestCheckResourceAttr(resourceName, "bid_price", ""),
+					resource.TestCheckResourceAttr(resourceName, "ebs_optimized", "false"),
+					resource.TestCheckResourceAttr(resourceName, "instance_role", emr.InstanceGroupTypeTask),
 				),
 			},
 			{
@@ -39,9 +43,63 @@ func TestAccAWSEMRInstanceGroup_basic(t *testing.T) {
 	})
 }
 
+func TestAccAWSEMRInstanceGroup_BidPrice(t *testing.T) {
+	var ig1, ig2 emr.InstanceGroup
+	rInt := acctest.RandInt()
+
+	resourceName := "aws_emr_instance_group.task"
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckAWSEmrInstanceGroupDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccAWSEmrInstanceGroupConfig_basic(rInt),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSEmrInstanceGroupExists(resourceName, &ig1),
+					resource.TestCheckResourceAttr(resourceName, "instance_role", emr.InstanceGroupTypeTask),
+					resource.TestCheckResourceAttr(resourceName, "bid_price", ""),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateIdFunc: testAccAWSEMRInstanceGroupResourceImportStateIdFunc(resourceName),
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccAWSEmrInstanceGroupConfig_BidPrice(rInt),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSEmrInstanceGroupExists(resourceName, &ig2),
+					resource.TestCheckResourceAttr(resourceName, "instance_role", emr.InstanceGroupTypeTask),
+					resource.TestCheckResourceAttr(resourceName, "bid_price", "0.30"),
+					testAccAWSEMRInstanceGroupRecreated(t, &ig1, &ig2),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateIdFunc: testAccAWSEMRInstanceGroupResourceImportStateIdFunc(resourceName),
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccAWSEmrInstanceGroupConfig_basic(rInt),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSEmrInstanceGroupExists(resourceName, &ig1),
+					resource.TestCheckResourceAttr(resourceName, "instance_role", emr.InstanceGroupTypeTask),
+					resource.TestCheckResourceAttr(resourceName, "bid_price", ""),
+					testAccAWSEMRInstanceGroupRecreated(t, &ig1, &ig2),
+				),
+			},
+		},
+	})
+}
+
 func TestAccAWSEMRInstanceGroup_AutoScalingPolicy(t *testing.T) {
 	var ig emr.InstanceGroup
 	rInt := acctest.RandInt()
+
+	resourceName := "aws_emr_instance_group.task"
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
 		Providers:    testAccProviders,
@@ -50,11 +108,17 @@ func TestAccAWSEMRInstanceGroup_AutoScalingPolicy(t *testing.T) {
 			{
 				Config: testAccAWSEmrInstanceGroupConfig_AutoScalingPolicy(rInt, 1, 3),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckAWSEmrInstanceGroupExists("aws_emr_instance_group.task", &ig),
-					resource.TestCheckResourceAttr("aws_emr_instance_group.task", "instance_role", "TASK"),
-					resource.TestCheckResourceAttrSet("aws_emr_instance_group.task", "autoscaling_policy"),
+					testAccCheckAWSEmrInstanceGroupExists(resourceName, &ig),
+					resource.TestCheckResourceAttr(resourceName, "instance_role", "TASK"),
+					resource.TestCheckResourceAttrSet(resourceName, "autoscaling_policy"),
 				),
 			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateIdFunc: testAccAWSEMRInstanceGroupResourceImportStateIdFunc(resourceName),
+				ImportStateVerify: true,
+			},
 		},
 	})
 }
@@ -62,6 +126,8 @@ func TestAccAWSEMRInstanceGroup_AutoScalingPolicy(t *testing.T) {
 func TestAccAWSEMRInstanceGroup_updateAutoScalingPolicy(t *testing.T) {
 	var ig emr.InstanceGroup
 	rInt := acctest.RandInt()
+
+	resourceName := "aws_emr_instance_group.task"
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
 		Providers:    testAccProviders,
@@ -70,17 +136,23 @@ func TestAccAWSEMRInstanceGroup_updateAutoScalingPolicy(t *testing.T) {
 			{
 				Config: testAccAWSEmrInstanceGroupConfig_AutoScalingPolicy(rInt, 1, 3),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckAWSEmrInstanceGroupExists("aws_emr_instance_group.task", &ig),
-					resource.TestCheckResourceAttr("aws_emr_instance_group.task", "instance_role", "TASK"),
-					resource.TestCheckResourceAttrSet("aws_emr_instance_group.task", "autoscaling_policy"),
+					testAccCheckAWSEmrInstanceGroupExists(resourceName, &ig),
+					resource.TestCheckResourceAttr(resourceName, "instance_role", "TASK"),
+					resource.TestCheckResourceAttrSet(resourceName, "autoscaling_policy"),
 				),
 			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateIdFunc: testAccAWSEMRInstanceGroupResourceImportStateIdFunc(resourceName),
+				ImportStateVerify: true,
+			},
 			{
 				Config: testAccAWSEmrInstanceGroupConfig_AutoScalingPolicy(rInt, 2, 3),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckAWSEmrInstanceGroupExists("aws_emr_instance_group.task", &ig),
-					resource.TestCheckResourceAttr("aws_emr_instance_group.task", "instance_role", "TASK"),
-					resource.TestCheckResourceAttrSet("aws_emr_instance_group.task", "autoscaling_policy"),
+					testAccCheckAWSEmrInstanceGroupExists(resourceName, &ig),
+					resource.TestCheckResourceAttr(resourceName, "instance_role", "TASK"),
+					resource.TestCheckResourceAttrSet(resourceName, "autoscaling_policy"),
 				),
 			},
 		},
@@ -92,6 +164,8 @@ func TestAccAWSEMRInstanceGroup_updateAutoScalingPolicy(t *testing.T) {
 func TestAccAWSEMRInstanceGroup_updateInstanceCount(t *testing.T) {
 	var ig emr.InstanceGroup
 	rInt := acctest.RandInt()
+
+	resourceName := "aws_emr_instance_group.task"
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
 		Providers:    testAccProviders,
@@ -99,11 +173,17 @@ func TestAccAWSEMRInstanceGroup_updateInstanceCount(t *testing.T) {
 		Steps: []resource.TestStep{
 			{
 				Config: testAccAWSEmrInstanceGroupConfig_basic(rInt),
-				Check:  testAccCheckAWSEmrInstanceGroupExists("aws_emr_instance_group.task", &ig),
+				Check:  testAccCheckAWSEmrInstanceGroupExists(resourceName, &ig),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateIdFunc: testAccAWSEMRInstanceGroupResourceImportStateIdFunc(resourceName),
+				ImportStateVerify: true,
 			},
 			{
 				Config: testAccAWSEmrInstanceGroupConfig_zeroCount(rInt),
-				Check:  testAccCheckAWSEmrInstanceGroupExists("aws_emr_instance_group.task", &ig),
+				Check:  testAccCheckAWSEmrInstanceGroupExists(resourceName, &ig),
 			},
 		},
 	})
@@ -112,6 +192,8 @@ func TestAccAWSEMRInstanceGroup_updateInstanceCount(t *testing.T) {
 func TestAccAWSEMRInstanceGroup_ebsConfig(t *testing.T) {
 	var ig emr.InstanceGroup
 	rInt := acctest.RandInt()
+
+	resourceName := "aws_emr_instance_group.task"
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
 		Providers:    testAccProviders,
@@ -120,11 +202,23 @@ func TestAccAWSEMRInstanceGroup_ebsConfig(t *testing.T) {
 			{
 				Config: testAccAWSEmrInstanceGroupConfig_ebsConfig(rInt),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckAWSEmrInstanceGroupExists("aws_emr_instance_group.task", &ig),
-					resource.TestCheckResourceAttr(
-						"aws_emr_instance_group.task", "ebs_config.#", "1"),
-					resource.TestCheckResourceAttr(
-						"aws_emr_instance_group.task", "ebs_optimized", "true"),
+					testAccCheckAWSEmrInstanceGroupExists(resourceName, &ig),
+					resource.TestCheckResourceAttr(resourceName, "ebs_config.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "ebs_optimized", "true"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateIdFunc: testAccAWSEMRInstanceGroupResourceImportStateIdFunc(resourceName),
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccAWSEmrInstanceGroupConfig_ebsConfig(rInt),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSEmrInstanceGroupExists(resourceName, &ig),
+					resource.TestCheckResourceAttr(resourceName, "ebs_config.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "ebs_optimized", "true"),
 				),
 			},
 		},
@@ -152,12 +246,10 @@ func testAccCheckAWSEmrInstanceGroupDestroy(s *terraform.State) error {
 			}
 		}
 
-		providerErr, ok := err.(awserr.Error)
-		if !ok {
+		if providerErr, ok := err.(awserr.Error); !ok {
+			log.Printf("[ERROR] %v", providerErr)
 			return err
 		}
-
-		log.Printf("[ERROR] %v", providerErr)
 	}
 
 	return nil
@@ -199,6 +291,15 @@ func testAccAWSEMRInstanceGroupResourceImportStateIdFunc(resourceName string) re
 	}
 }
 
+func testAccAWSEMRInstanceGroupRecreated(t *testing.T, before, after *emr.InstanceGroup) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		if *before.Id == *after.Id {
+			t.Fatalf("Expected change of Instance Group Ids, but both were %v", before.Id)
+		}
+		return nil
+	}
+}
+
 const testAccAWSEmrInstanceGroupBase = `
 resource "aws_security_group" "allow_all" {
   name        = "allow_all"
@@ -484,6 +585,17 @@ func testAccAWSEmrInstanceGroupConfig_basic(r int) string {
 	`, r)
 }
 
+func testAccAWSEmrInstanceGroupConfig_BidPrice(r int) string {
+	return fmt.Sprintf(testAccAWSEmrInstanceGroupBase+`
+	resource "aws_emr_instance_group" "task" {
+    cluster_id     = "${aws_emr_cluster.tf-test-cluster.id}"
+    bid_price			 = "0.30"
+    instance_count = 1
+    instance_type  = "c4.large"
+  }
+	`, r)
+}
+
 func testAccAWSEmrInstanceGroupConfig_AutoScalingPolicy(r, min, max int) string {
 	return fmt.Sprintf(testAccAWSEmrInstanceGroupBase+`
 	resource "aws_emr_instance_group" "task" {
diff --git a/website/docs/r/emr_instance_group.html.markdown b/website/docs/r/emr_instance_group.html.markdown
index fe64ace0a147..dc3f5c57dcf5 100644
--- a/website/docs/r/emr_instance_group.html.markdown
+++ b/website/docs/r/emr_instance_group.html.markdown
@@ -32,7 +32,7 @@ The following arguments are supported:
 
 * `name` (Required) Human friendly name given to the instance group. Changing this forces a new resource to be created.
 * `cluster_id` (Required) ID of the EMR Cluster to attach to. Changing this forces a new resource to be created.
-* `instance_role` (Optional) The EC2 instance role for all instances in the instance group. Valid values are CORE, TASK. Default `"TASK"`
+* `instance_role` (Optional) The EC2 instance role for all instances in the instance group. Default `"TASK"`
 * `instance_type` (Required) The EC2 instance type for all instances in the instance group. Changing this forces a new resource to be created.
 * `instance_count` (optional) target number of instances for the instance group. defaults to 0.
 * `bid_price` - (Optional) If set, the bid price for each EC2 instance in the instance group, expressed in USD. By setting this attribute, the instance group is being declared as a Spot Instance, and will implicitly create a Spot request. Leave this blank to use On-Demand Instances.