diff --git a/internal/command/views/apply.go b/internal/command/views/apply.go
index a5fb82d24f..8a8dbc674d 100644
--- a/internal/command/views/apply.go
+++ b/internal/command/views/apply.go
@@ -66,6 +66,14 @@ func (v *ApplyHuman) ResourceCount(stateOutPath string) {
 			v.view.colorize.Color("[reset][bold][green]\nDestroy complete! Resources: %d destroyed.\n"),
 			v.countHook.Removed,
 		)
+	} else if v.countHook.Imported > 0 {
+		v.view.streams.Printf(
+			v.view.colorize.Color("[reset][bold][green]\nApply complete! Resources: %d imported, %d added, %d changed, %d destroyed.\n"),
+			v.countHook.Imported,
+			v.countHook.Added,
+			v.countHook.Changed,
+			v.countHook.Removed,
+		)
 	} else {
 		v.view.streams.Printf(
 			v.view.colorize.Color("[reset][bold][green]\nApply complete! Resources: %d added, %d changed, %d destroyed.\n"),
@@ -133,6 +141,7 @@ func (v *ApplyJSON) ResourceCount(stateOutPath string) {
 		Add:       v.countHook.Added,
 		Change:    v.countHook.Changed,
 		Remove:    v.countHook.Removed,
+		Import:    v.countHook.Imported,
 		Operation: operation,
 	})
 }
diff --git a/internal/command/views/apply_test.go b/internal/command/views/apply_test.go
index 4c8f92d97a..dfaa46caab 100644
--- a/internal/command/views/apply_test.go
+++ b/internal/command/views/apply_test.go
@@ -103,16 +103,24 @@ func TestApplyHuman_help(t *testing.T) {
 // Hooks and ResourceCount are tangled up and easiest to test together.
 func TestApply_resourceCount(t *testing.T) {
 	testCases := map[string]struct {
-		destroy bool
-		want    string
+		destroy   bool
+		want      string
+		importing bool
 	}{
 		"apply": {
 			false,
 			"Apply complete! Resources: 1 added, 2 changed, 3 destroyed.",
+			false,
 		},
 		"destroy": {
 			true,
 			"Destroy complete! Resources: 3 destroyed.",
+			false,
+		},
+		"import": {
+			false,
+			"Apply complete! Resources: 1 imported, 1 added, 2 changed, 3 destroyed.",
+			true,
 		},
 	}
 
@@ -141,6 +149,10 @@ func TestApply_resourceCount(t *testing.T) {
 				count.Changed = 2
 				count.Removed = 3
 
+				if tc.importing {
+					count.Imported = 1
+				}
+
 				v.ResourceCount("")
 
 				got := done(t).Stdout()
diff --git a/internal/command/views/hook_count.go b/internal/command/views/hook_count.go
index 3f740647a5..2336ee61d3 100644
--- a/internal/command/views/hook_count.go
+++ b/internal/command/views/hook_count.go
@@ -17,9 +17,10 @@ import (
 // countHook is a hook that counts the number of resources
 // added, removed, changed during the course of an apply.
 type countHook struct {
-	Added   int
-	Changed int
-	Removed int
+	Added    int
+	Changed  int
+	Removed  int
+	Imported int
 
 	ToAdd          int
 	ToChange       int
@@ -42,6 +43,7 @@ func (h *countHook) Reset() {
 	h.Added = 0
 	h.Changed = 0
 	h.Removed = 0
+	h.Imported = 0
 }
 
 func (h *countHook) PreApply(addr addrs.AbsResourceInstance, gen states.Generation, action plans.Action, priorState, plannedNewState cty.Value) (terraform.HookAction, error) {
@@ -107,3 +109,11 @@ func (h *countHook) PostDiff(addr addrs.AbsResourceInstance, gen states.Generati
 
 	return terraform.HookActionContinue, nil
 }
+
+func (h *countHook) PostApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (terraform.HookAction, error) {
+	h.Lock()
+	defer h.Unlock()
+
+	h.Imported++
+	return terraform.HookActionContinue, nil
+}
diff --git a/internal/command/views/hook_ui.go b/internal/command/views/hook_ui.go
index 491a51946d..400718eed1 100644
--- a/internal/command/views/hook_ui.go
+++ b/internal/command/views/hook_ui.go
@@ -304,6 +304,33 @@ func (h *UiHook) PostImportState(addr addrs.AbsResourceInstance, imported []prov
 	return terraform.HookActionContinue, nil
 }
 
+func (h *UiHook) PrePlanImport(addr addrs.AbsResourceInstance, importID string) (terraform.HookAction, error) {
+	h.println(fmt.Sprintf(
+		h.view.colorize.Color("[reset][bold]%s: Preparing import... [id=%s]"),
+		addr, importID,
+	))
+
+	return terraform.HookActionContinue, nil
+}
+
+func (h *UiHook) PreApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (terraform.HookAction, error) {
+	h.println(fmt.Sprintf(
+		h.view.colorize.Color("[reset][bold]%s: Importing... [id=%s]"),
+		addr, importing.ID,
+	))
+
+	return terraform.HookActionContinue, nil
+}
+
+func (h *UiHook) PostApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (terraform.HookAction, error) {
+	h.println(fmt.Sprintf(
+		h.view.colorize.Color("[reset][bold]%s: Import complete [id=%s]"),
+		addr, importing.ID,
+	))
+
+	return terraform.HookActionContinue, nil
+}
+
 // Wrap calls to the view so that concurrent calls do not interleave println.
 func (h *UiHook) println(s string) {
 	h.viewLock.Lock()
diff --git a/internal/command/views/json/change_summary.go b/internal/command/views/json/change_summary.go
index a987b2b95a..3915c5c1fd 100644
--- a/internal/command/views/json/change_summary.go
+++ b/internal/command/views/json/change_summary.go
@@ -25,18 +25,11 @@ type ChangeSummary struct {
 // used by Terraform Cloud and Terraform Enterprise, so the exact formats of
 // these strings are important.
 func (cs *ChangeSummary) String() string {
-
-	// TODO(liamcervante): For now, we only include the import count in the plan
-	//   output. This is because counting the imports during the apply is tricky
-	//   and we need to use the actual implementation which isn't ready yet.
-	//
-	//   We should absolutely fix this before we launch to alpha, but we can't
-	//   do it right now. So we have implemented as much as we can (the plan)
-	//   and will revisit this alongside the concrete implementation of the
-	//   Terraform graph.
-
 	switch cs.Operation {
 	case OperationApplied:
+		if cs.Import > 0 {
+			return fmt.Sprintf("Apply complete! Resources: %d imported, %d added, %d changed, %d destroyed.", cs.Import, cs.Add, cs.Change, cs.Remove)
+		}
 		return fmt.Sprintf("Apply complete! Resources: %d added, %d changed, %d destroyed.", cs.Add, cs.Change, cs.Remove)
 	case OperationDestroyed:
 		return fmt.Sprintf("Destroy complete! Resources: %d destroyed.", cs.Remove)
diff --git a/internal/command/views/json_view_test.go b/internal/command/views/json_view_test.go
index 209e4f955d..61a6a38182 100644
--- a/internal/command/views/json_view_test.go
+++ b/internal/command/views/json_view_test.go
@@ -215,6 +215,36 @@ func TestJSONView_ChangeSummary(t *testing.T) {
 	testJSONViewOutputEquals(t, done(t).Stdout(), want)
 }
 
+func TestJSONView_ChangeSummaryWithImport(t *testing.T) {
+	streams, done := terminal.StreamsForTesting(t)
+	jv := NewJSONView(NewView(streams))
+
+	jv.ChangeSummary(&viewsjson.ChangeSummary{
+		Add:       1,
+		Change:    2,
+		Remove:    3,
+		Import:    1,
+		Operation: viewsjson.OperationApplied,
+	})
+
+	want := []map[string]interface{}{
+		{
+			"@level":   "info",
+			"@message": "Apply complete! Resources: 1 imported, 1 added, 2 changed, 3 destroyed.",
+			"@module":  "terraform.ui",
+			"type":     "change_summary",
+			"changes": map[string]interface{}{
+				"add":       float64(1),
+				"change":    float64(2),
+				"remove":    float64(3),
+				"import":    float64(1),
+				"operation": "apply",
+			},
+		},
+	}
+	testJSONViewOutputEquals(t, done(t).Stdout(), want)
+}
+
 func TestJSONView_Hook(t *testing.T) {
 	streams, done := terminal.StreamsForTesting(t)
 	jv := NewJSONView(NewView(streams))
diff --git a/internal/terraform/context_apply.go b/internal/terraform/context_apply.go
index e1f3c951d8..4f5851fab3 100644
--- a/internal/terraform/context_apply.go
+++ b/internal/terraform/context_apply.go
@@ -38,6 +38,19 @@ func (c *Context) Apply(plan *plans.Plan, config *configs.Config) (*states.State
 		return nil, diags
 	}
 
+	for _, rc := range plan.Changes.Resources {
+		// Import is a no-op change during an apply (all the real action happens during the plan) but we'd
+		// like to show some helpful output that mirrors the way we show other changes.
+		if rc.Importing != nil {
+			for _, h := range c.hooks {
+				// In future, we may need to call PostApplyImport separately elsewhere in the apply
+				// operation. For now, though, we'll call Pre and Post hooks together.
+				h.PreApplyImport(rc.Addr, *rc.Importing)
+				h.PostApplyImport(rc.Addr, *rc.Importing)
+			}
+		}
+	}
+
 	graph, operation, diags := c.applyGraph(plan, config, true)
 	if diags.HasErrors() {
 		return nil, diags
diff --git a/internal/terraform/context_apply2_test.go b/internal/terraform/context_apply2_test.go
index ab6269a08a..9442c22899 100644
--- a/internal/terraform/context_apply2_test.go
+++ b/internal/terraform/context_apply2_test.go
@@ -2088,3 +2088,77 @@ resource "unused_resource" "test" {
 	_, diags = ctx.Apply(plan, m)
 	assertNoErrors(t, diags)
 }
+
+func TestContext2Apply_import(t *testing.T) {
+	m := testModuleInline(t, map[string]string{
+		"main.tf": `
+resource "test_resource" "a" {
+  id = "importable"
+}
+
+import {
+  to = test_resource.a
+  id = "importable" 
+}
+`,
+	})
+
+	p := testProvider("test")
+	p.GetProviderSchemaResponse = getProviderSchemaResponseFromProviderSchema(&ProviderSchema{
+		ResourceTypes: map[string]*configschema.Block{
+			"test_resource": {
+				Attributes: map[string]*configschema.Attribute{
+					"id": {
+						Type:     cty.String,
+						Required: true,
+					},
+				},
+			},
+		},
+	})
+	p.PlanResourceChangeFn = func(req providers.PlanResourceChangeRequest) providers.PlanResourceChangeResponse {
+		return providers.PlanResourceChangeResponse{
+			PlannedState: req.ProposedNewState,
+		}
+	}
+	p.ImportResourceStateFn = func(req providers.ImportResourceStateRequest) providers.ImportResourceStateResponse {
+		return providers.ImportResourceStateResponse{
+			ImportedResources: []providers.ImportedResource{
+				{
+					TypeName: "test_instance",
+					State: cty.ObjectVal(map[string]cty.Value{
+						"id": cty.StringVal("importable"),
+					}),
+				},
+			},
+		}
+	}
+	hook := new(MockHook)
+	ctx := testContext2(t, &ContextOpts{
+		Hooks: []Hook{hook},
+		Providers: map[addrs.Provider]providers.Factory{
+			addrs.NewDefaultProvider("test"): testProviderFuncFixed(p),
+		},
+	})
+	plan, diags := ctx.Plan(m, states.NewState(), &PlanOpts{
+		Mode: plans.NormalMode,
+	})
+	assertNoErrors(t, diags)
+
+	_, diags = ctx.Apply(plan, m)
+	assertNoErrors(t, diags)
+
+	if !hook.PreApplyImportCalled {
+		t.Fatalf("PreApplyImport hook not called")
+	}
+	if addr, wantAddr := hook.PreApplyImportAddr, mustResourceInstanceAddr("test_resource.a"); !addr.Equal(wantAddr) {
+		t.Errorf("expected addr to be %s, but was %s", wantAddr, addr)
+	}
+
+	if !hook.PostApplyImportCalled {
+		t.Fatalf("PostApplyImport hook not called")
+	}
+	if addr, wantAddr := hook.PostApplyImportAddr, mustResourceInstanceAddr("test_resource.a"); !addr.Equal(wantAddr) {
+		t.Errorf("expected addr to be %s, but was %s", wantAddr, addr)
+	}
+}
diff --git a/internal/terraform/context_plan2_test.go b/internal/terraform/context_plan2_test.go
index 765a45f7a0..f4dd5bcc86 100644
--- a/internal/terraform/context_plan2_test.go
+++ b/internal/terraform/context_plan2_test.go
@@ -4118,7 +4118,9 @@ import {
 	})
 
 	p := simpleMockProvider()
+	hook := new(MockHook)
 	ctx := testContext2(t, &ContextOpts{
+		Hooks: []Hook{hook},
 		Providers: map[addrs.Provider]providers.Factory{
 			addrs.NewDefaultProvider("test"): testProviderFuncFixed(p),
 		},
@@ -4165,6 +4167,20 @@ import {
 		if instPlan.Importing.ID != "123" {
 			t.Errorf("expected import change from \"123\", got non-import change")
 		}
+
+		if !hook.PrePlanImportCalled {
+			t.Fatalf("PostPlanImport hook not called")
+		}
+		if addr, wantAddr := hook.PrePlanImportAddr, instPlan.Addr; !addr.Equal(wantAddr) {
+			t.Errorf("expected addr to be %s, but was %s", wantAddr, addr)
+		}
+
+		if !hook.PostPlanImportCalled {
+			t.Fatalf("PostPlanImport hook not called")
+		}
+		if addr, wantAddr := hook.PostPlanImportAddr, instPlan.Addr; !addr.Equal(wantAddr) {
+			t.Errorf("expected addr to be %s, but was %s", wantAddr, addr)
+		}
 	})
 }
 
diff --git a/internal/terraform/hook.go b/internal/terraform/hook.go
index f6d99b650b..f23247c50a 100644
--- a/internal/terraform/hook.go
+++ b/internal/terraform/hook.go
@@ -75,10 +75,21 @@ type Hook interface {
 	PostRefresh(addr addrs.AbsResourceInstance, gen states.Generation, priorState cty.Value, newState cty.Value) (HookAction, error)
 
 	// PreImportState and PostImportState are called before and after
-	// (respectively) each state import operation for a given resource address.
+	// (respectively) each state import operation for a given resource address when
+	// using the legacy import command.
 	PreImportState(addr addrs.AbsResourceInstance, importID string) (HookAction, error)
 	PostImportState(addr addrs.AbsResourceInstance, imported []providers.ImportedResource) (HookAction, error)
 
+	// PrePlanImport and PostPlanImport are called during a plan before and after planning to import
+	// a new resource using the configuration-driven import workflow.
+	PrePlanImport(addr addrs.AbsResourceInstance, importID string) (HookAction, error)
+	PostPlanImport(addr addrs.AbsResourceInstance, imported []providers.ImportedResource) (HookAction, error)
+
+	// PreApplyImport and PostApplyImport are called during an apply for each imported resource when
+	// using the configuration-driven import workflow.
+	PreApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (HookAction, error)
+	PostApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (HookAction, error)
+
 	// Stopping is called if an external signal requests that Terraform
 	// gracefully abort an operation in progress.
 	//
@@ -159,6 +170,22 @@ func (*NilHook) PostImportState(addr addrs.AbsResourceInstance, imported []provi
 	return HookActionContinue, nil
 }
 
+func (h *NilHook) PrePlanImport(addr addrs.AbsResourceInstance, importID string) (HookAction, error) {
+	return HookActionContinue, nil
+}
+
+func (h *NilHook) PostPlanImport(addr addrs.AbsResourceInstance, imported []providers.ImportedResource) (HookAction, error) {
+	return HookActionContinue, nil
+}
+
+func (h *NilHook) PreApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (HookAction, error) {
+	return HookActionContinue, nil
+}
+
+func (h *NilHook) PostApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (HookAction, error) {
+	return HookActionContinue, nil
+}
+
 func (*NilHook) Stopping() {
 	// Does nothing at all by default
 }
diff --git a/internal/terraform/hook_mock.go b/internal/terraform/hook_mock.go
index 786e3936d4..ae9aa39683 100644
--- a/internal/terraform/hook_mock.go
+++ b/internal/terraform/hook_mock.go
@@ -111,6 +111,26 @@ type MockHook struct {
 	PostImportStateReturn    HookAction
 	PostImportStateError     error
 
+	PrePlanImportCalled bool
+	PrePlanImportAddr   addrs.AbsResourceInstance
+	PrePlanImportReturn HookAction
+	PrePlanImportError  error
+
+	PostPlanImportAddr   addrs.AbsResourceInstance
+	PostPlanImportCalled bool
+	PostPlanImportReturn HookAction
+	PostPlanImportError  error
+
+	PreApplyImportCalled bool
+	PreApplyImportAddr   addrs.AbsResourceInstance
+	PreApplyImportReturn HookAction
+	PreApplyImportError  error
+
+	PostApplyImportCalled bool
+	PostApplyImportAddr   addrs.AbsResourceInstance
+	PostApplyImportReturn HookAction
+	PostApplyImportError  error
+
 	StoppingCalled bool
 
 	PostStateUpdateCalled bool
@@ -269,6 +289,33 @@ func (h *MockHook) PostImportState(addr addrs.AbsResourceInstance, imported []pr
 	return h.PostImportStateReturn, h.PostImportStateError
 }
 
+func (h *MockHook) PrePlanImport(addr addrs.AbsResourceInstance, importID string) (HookAction, error) {
+	h.PrePlanImportCalled = true
+	h.PrePlanImportAddr = addr
+	return h.PrePlanImportReturn, h.PrePlanImportError
+}
+
+func (h *MockHook) PostPlanImport(addr addrs.AbsResourceInstance, imported []providers.ImportedResource) (HookAction, error) {
+	h.PostPlanImportCalled = true
+	h.PostPlanImportAddr = addr
+	return h.PostPlanImportReturn, h.PostPlanImportError
+}
+
+func (h *MockHook) PreApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (HookAction, error) {
+	h.PreApplyImportCalled = true
+	h.PreApplyImportAddr = addr
+	return h.PreApplyImportReturn, h.PreApplyImportError
+}
+
+func (h *MockHook) PostApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (HookAction, error) {
+	h.Lock()
+	defer h.Unlock()
+
+	h.PostApplyImportCalled = true
+	h.PostApplyImportAddr = addr
+	return h.PostApplyImportReturn, h.PostApplyImportError
+}
+
 func (h *MockHook) Stopping() {
 	h.Lock()
 	defer h.Unlock()
diff --git a/internal/terraform/hook_stop.go b/internal/terraform/hook_stop.go
index 510a2ee785..52ead5ac4e 100644
--- a/internal/terraform/hook_stop.go
+++ b/internal/terraform/hook_stop.go
@@ -74,6 +74,22 @@ func (h *stopHook) PostImportState(addr addrs.AbsResourceInstance, imported []pr
 	return h.hook()
 }
 
+func (h *stopHook) PrePlanImport(addr addrs.AbsResourceInstance, importID string) (HookAction, error) {
+	return h.hook()
+}
+
+func (h *stopHook) PostPlanImport(addr addrs.AbsResourceInstance, imported []providers.ImportedResource) (HookAction, error) {
+	return h.hook()
+}
+
+func (h *stopHook) PreApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (HookAction, error) {
+	return h.hook()
+}
+
+func (h *stopHook) PostApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (HookAction, error) {
+	return h.hook()
+}
+
 func (h *stopHook) Stopping() {}
 
 func (h *stopHook) PostStateUpdate(new *states.State) (HookAction, error) {
diff --git a/internal/terraform/hook_test.go b/internal/terraform/hook_test.go
index f1225a2e7a..e0c845535a 100644
--- a/internal/terraform/hook_test.go
+++ b/internal/terraform/hook_test.go
@@ -127,6 +127,34 @@ func (h *testHook) PostImportState(addr addrs.AbsResourceInstance, imported []pr
 	return HookActionContinue, nil
 }
 
+func (h *testHook) PrePlanImport(addr addrs.AbsResourceInstance, importID string) (HookAction, error) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.Calls = append(h.Calls, &testHookCall{"PrePlanImport", addr.String()})
+	return HookActionContinue, nil
+}
+
+func (h *testHook) PostPlanImport(addr addrs.AbsResourceInstance, imported []providers.ImportedResource) (HookAction, error) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.Calls = append(h.Calls, &testHookCall{"PostPlanImport", addr.String()})
+	return HookActionContinue, nil
+}
+
+func (h *testHook) PreApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (HookAction, error) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.Calls = append(h.Calls, &testHookCall{"PreApplyImport", addr.String()})
+	return HookActionContinue, nil
+}
+
+func (h *testHook) PostApplyImport(addr addrs.AbsResourceInstance, importing plans.ImportingSrc) (HookAction, error) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.Calls = append(h.Calls, &testHookCall{"PostApplyImport", addr.String()})
+	return HookActionContinue, nil
+}
+
 func (h *testHook) Stopping() {
 	h.mu.Lock()
 	defer h.mu.Unlock()
diff --git a/internal/terraform/node_resource_plan_instance.go b/internal/terraform/node_resource_plan_instance.go
index ce9c709d26..040739b463 100644
--- a/internal/terraform/node_resource_plan_instance.go
+++ b/internal/terraform/node_resource_plan_instance.go
@@ -400,7 +400,7 @@ func (n *NodePlannableResourceInstance) importState(ctx EvalContext, addr addrs.
 	absAddr := addr.Resource.Absolute(ctx.Path())
 
 	diags = diags.Append(ctx.Hook(func(h Hook) (HookAction, error) {
-		return h.PreImportState(absAddr, n.importTarget.ID)
+		return h.PrePlanImport(absAddr, n.importTarget.ID)
 	}))
 	if diags.HasErrors() {
 		return nil, diags
@@ -446,7 +446,7 @@ func (n *NodePlannableResourceInstance) importState(ctx EvalContext, addr addrs.
 
 	// call post-import hook
 	diags = diags.Append(ctx.Hook(func(h Hook) (HookAction, error) {
-		return h.PostImportState(absAddr, imported)
+		return h.PostPlanImport(absAddr, imported)
 	}))
 
 	if imported[0].TypeName == "" {