diff --git a/.changelog/3375.txt b/.changelog/3375.txt
new file mode 100644
index 00000000000..cc742cc9a72
--- /dev/null
+++ b/.changelog/3375.txt
@@ -0,0 +1,3 @@
+```release-note:bug
+dataproc: fixed diff when `google_dataproc_cluster` `preemptible_worker_config.0.num_instances` is sized to 0 and other `preemptible_worker_config` subfields are set
+```
diff --git a/google/resource_dataproc_cluster.go b/google/resource_dataproc_cluster.go
index 6b899b24eac..673f283eab5 100644
--- a/google/resource_dataproc_cluster.go
+++ b/google/resource_dataproc_cluster.go
@@ -1269,6 +1269,22 @@ func flattenGceClusterConfig(d *schema.ResourceData, gcc *dataproc.GceClusterCon
 }
 
 func flattenPreemptibleInstanceGroupConfig(d *schema.ResourceData, icg *dataproc.InstanceGroupConfig) []map[string]interface{} {
+	// if num_instances is 0, icg will always be returned nil. This means the
+	// server has discarded diskconfig etc. However, the only way to remove the
+	// preemptible group is to set the size to 0, because it's O+C. Many users
+	// won't remove the rest of the config (eg disk config). Therefore, we need to
+	// preserve the other set fields by using the old state to stop users from
+	// getting a diff.
+	if icg == nil {
+		icgSchema := d.Get("cluster_config.0.preemptible_worker_config")
+		log.Printf("[DEBUG] state of preemptible is %#v", icgSchema)
+		if v, ok := icgSchema.([]interface{}); ok && len(v) > 0 {
+			if m, ok := v[0].(map[string]interface{}); ok {
+				return []map[string]interface{}{m}
+			}
+		}
+	}
+
 	disk := map[string]interface{}{}
 	data := map[string]interface{}{}
 
diff --git a/google/resource_dataproc_cluster_test.go b/google/resource_dataproc_cluster_test.go
index b45f0d059f8..fa7090f0fb6 100644
--- a/google/resource_dataproc_cluster_test.go
+++ b/google/resource_dataproc_cluster_test.go
@@ -369,6 +369,14 @@ func TestAccDataprocCluster_updatable(t *testing.T) {
 					resource.TestCheckResourceAttr("google_dataproc_cluster.updatable", "cluster_config.0.worker_config.0.num_instances", "2"),
 					resource.TestCheckResourceAttr("google_dataproc_cluster.updatable", "cluster_config.0.preemptible_worker_config.0.num_instances", "1")),
 			},
+			{
+				Config: testAccDataprocCluster_updatable(rnd, 2, 0),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckDataprocClusterExists("google_dataproc_cluster.updatable", &cluster),
+					resource.TestCheckResourceAttr("google_dataproc_cluster.updatable", "cluster_config.0.master_config.0.num_instances", "1"),
+					resource.TestCheckResourceAttr("google_dataproc_cluster.updatable", "cluster_config.0.worker_config.0.num_instances", "2"),
+					resource.TestCheckResourceAttr("google_dataproc_cluster.updatable", "cluster_config.0.preemptible_worker_config.0.num_instances", "0")),
+			},
 			{
 				Config: testAccDataprocCluster_updatable(rnd, 3, 2),
 				Check: resource.ComposeTestCheckFunc(