diff --git a/beets/config_default.yaml b/beets/config_default.yaml
index b42594f87..66ab9526f 100644
--- a/beets/config_default.yaml
+++ b/beets/config_default.yaml
@@ -19,6 +19,7 @@ import:
     languages: []
     detail: no
     flat: no
+    group_albums: no
 
 clutter: ["Thumbs.DB", ".DS_Store"]
 ignore: [".*", "*~", "System Volume Information"]
diff --git a/beets/importer.py b/beets/importer.py
index 9e3e7f193..28d94eeaa 100644
--- a/beets/importer.py
+++ b/beets/importer.py
@@ -325,6 +325,9 @@ class ImportSession(object):
                 stages += [item_progress(self)]
         else:
             # Whole-album importer.
+            if config['import']['group_albums']:
+                # Split directory tasks into one task for each album
+                stages += [group_albums(self)]
             if config['import']['autotag']:
                 # Only look up and query the user when autotagging.
                 stages += [initial_lookup(self), user_query(self)]
@@ -1006,3 +1009,23 @@ def item_progress(session):
         log.info(displayable_path(task.item.path))
         task.set_null_candidates()
         task.set_choice(action.ASIS)
+
+
+def group_albums(session):
+    """Group the items of a task by albumartist and album name and create a new
+    task for each album. Yield the tasks as a multi message.
+    """
+    def group(item):
+        return (item.albumartist or item.artist, item.album)
+
+    task = None
+    while True:
+        task = yield task
+        if task.sentinel:
+            continue
+        tasks = []
+        for _, items in itertools.groupby(task.items, group):
+            tasks.append(ImportTask(items=list(items)))
+        tasks.append(ImportTask.progress_sentinel(task.toppath, task.paths))
+
+        task = pipeline.multiple(tasks)
diff --git a/test/test_importer.py b/test/test_importer.py
index c1682b3d7..6f48de2c5 100644
--- a/test/test_importer.py
+++ b/test/test_importer.py
@@ -518,9 +518,9 @@ class ImportExistingTest(_common.TestCase, ImportHelper):
         self.importer.run()
         self.assertNotExists(self.import_media[0].path)
 
-class ImportFlatAlbumTest(_common.TestCase, ImportHelper):
+class GroupAlbumsImportTest(_common.TestCase, ImportHelper):
     def setUp(self):
-        super(ImportFlatAlbumTest, self).setUp()
+        super(GroupAlbumsImportTest, self).setUp()
         self._setup_library()
         self._create_import_dir(3)
 
@@ -571,6 +571,14 @@ class ImportFlatAlbumTest(_common.TestCase, ImportHelper):
         artists = set([album.albumartist for album in self.lib.albums()])
         self.assertEqual(artists, set(['Artist B', 'Tag Artist']))
 
+class GlobalGroupAlbumsImportTest(GroupAlbumsImportTest):
+
+    def setUp(self):
+        super(GlobalGroupAlbumsImportTest, self).setUp()
+        self.importer.choice = importer.action.ASIS
+        config['import']['group_albums'] = True
+
+
 class InferAlbumDataTest(_common.TestCase):
     def setUp(self):
         super(InferAlbumDataTest, self).setUp()