diff --git a/compose/cli/main.py b/compose/cli/main.py
index e941c005c..a2375516e 100644
--- a/compose/cli/main.py
+++ b/compose/cli/main.py
@@ -317,10 +317,7 @@ class TopLevelCommand(Command):
         }
 
         if options['-e']:
-            # Merge environment from config with -e command line
-            container_options['environment'] = dict(
-                parse_environment(service.options.get('environment')),
-                **parse_environment(options['-e']))
+            container_options['environment'] = parse_environment(options['-e'])
 
         if options['--entrypoint']:
             container_options['entrypoint'] = options.get('--entrypoint')
diff --git a/compose/container.py b/compose/container.py
index fc3370d9e..6388ca80c 100644
--- a/compose/container.py
+++ b/compose/container.py
@@ -44,6 +44,10 @@ class Container(object):
     def image(self):
         return self.dictionary['Image']
 
+    @property
+    def image_config(self):
+        return self.client.inspect_image(self.image)
+
     @property
     def short_id(self):
         return self.id[:10]
diff --git a/compose/service.py b/compose/service.py
index 20f8db0a4..08d203274 100644
--- a/compose/service.py
+++ b/compose/service.py
@@ -3,14 +3,14 @@ from __future__ import absolute_import
 from collections import namedtuple
 import logging
 import re
-from operator import attrgetter
 import sys
-import six
+from operator import attrgetter
 
+import six
 from docker.errors import APIError
 from docker.utils import create_host_config, LogConfig
 
-from .config import DOCKER_CONFIG_KEYS
+from .config import DOCKER_CONFIG_KEYS, merge_environment
 from .container import Container, get_container_name
 from .progress_stream import stream_output, StreamOutputError
 
@@ -183,10 +183,10 @@ class Service(object):
         Create a container for this service. If the image doesn't exist, attempt to pull
         it.
         """
-        override_options['volumes_from'] = self._get_volumes_from(previous_container)
         container_options = self._get_container_create_options(
             override_options,
             one_off=one_off,
+            previous_container=previous_container,
         )
 
         if (do_build and
@@ -247,6 +247,12 @@ class Service(object):
         self.client.rename(
             container.id,
             '%s_%s' % (container.short_id, container.name))
+
+        override_options = dict(
+            override_options,
+            environment=merge_environment(
+                override_options.get('environment'),
+                {'affinity:container': '=' + container.id}))
         new_container = self.create_container(
             do_build=False,
             previous_container=container,
@@ -326,7 +332,7 @@ class Service(object):
             links.append((external_link, link_name))
         return links
 
-    def _get_volumes_from(self, previous_container=None):
+    def _get_volumes_from(self):
         volumes_from = []
         for volume_source in self.volumes_from:
             if isinstance(volume_source, Service):
@@ -339,9 +345,6 @@ class Service(object):
             elif isinstance(volume_source, Container):
                 volumes_from.append(volume_source.id)
 
-        if previous_container:
-            volumes_from.append(previous_container.id)
-
         return volumes_from
 
     def _get_net(self):
@@ -363,7 +366,11 @@ class Service(object):
 
         return net
 
-    def _get_container_create_options(self, override_options, one_off=False):
+    def _get_container_create_options(
+            self,
+            override_options,
+            one_off=False,
+            previous_container=None):
         container_options = dict(
             (k, self.options[k])
             for k in DOCKER_CONFIG_KEYS if k in self.options)
@@ -396,11 +403,19 @@ class Service(object):
                 ports.append(port)
             container_options['ports'] = ports
 
+        override_options['binds'] = merge_volume_bindings(
+            container_options.get('volumes') or [],
+            previous_container)
+
         if 'volumes' in container_options:
             container_options['volumes'] = dict(
                 (parse_volume_spec(v).internal, {})
                 for v in container_options['volumes'])
 
+        container_options['environment'] = merge_environment(
+            self.options.get('environment'),
+            override_options.get('environment'))
+
         if self.can_be_built():
             container_options['image'] = self.full_name
 
@@ -418,11 +433,6 @@ class Service(object):
         options = dict(self.options, **override_options)
         port_bindings = build_port_bindings(options.get('ports') or [])
 
-        volume_bindings = dict(
-            build_volume_binding(parse_volume_spec(volume))
-            for volume in options.get('volumes') or []
-            if ':' in volume)
-
         privileged = options.get('privileged', False)
         cap_add = options.get('cap_add', None)
         cap_drop = options.get('cap_drop', None)
@@ -447,8 +457,8 @@ class Service(object):
         return create_host_config(
             links=self._get_links(link_to_self=one_off),
             port_bindings=port_bindings,
-            binds=volume_bindings,
-            volumes_from=options.get('volumes_from'),
+            binds=options.get('binds'),
+            volumes_from=self._get_volumes_from(),
             privileged=privileged,
             network_mode=self._get_net(),
             devices=devices,
@@ -531,6 +541,50 @@ class Service(object):
         stream_output(output, sys.stdout)
 
 
+def get_container_data_volumes(container, volumes_option):
+    """Find the container data volumes that are in `volumes_option`, and return
+    a mapping of volume bindings for those volumes.
+    """
+    volumes = []
+
+    volumes_option = volumes_option or []
+    container_volumes = container.get('Volumes') or {}
+    image_volumes = container.image_config['ContainerConfig'].get('Volumes') or {}
+
+    for volume in set(volumes_option + image_volumes.keys()):
+        volume = parse_volume_spec(volume)
+        # No need to preserve host volumes
+        if volume.external:
+            continue
+
+        volume_path = container_volumes.get(volume.internal)
+        # New volume, doesn't exist in the old container
+        if not volume_path:
+            continue
+
+        # Copy existing volume from old container
+        volume = volume._replace(external=volume_path)
+        volumes.append(build_volume_binding(volume))
+
+    return dict(volumes)
+
+
+def merge_volume_bindings(volumes_option, previous_container):
+    """Return a list of volume bindings for a container. Container data volumes
+    are replaced by those from the previous container.
+    """
+    volume_bindings = dict(
+        build_volume_binding(parse_volume_spec(volume))
+        for volume in volumes_option or []
+        if ':' in volume)
+
+    if previous_container:
+        volume_bindings.update(
+            get_container_data_volumes(previous_container, volumes_option))
+
+    return volume_bindings
+
+
 NAME_RE = re.compile(r'^([^_]+)_([^_]+)_(run_)?(\d+)$')
 
 
diff --git a/tests/integration/service_test.py b/tests/integration/service_test.py
index 08e92a57f..bc21ab018 100644
--- a/tests/integration/service_test.py
+++ b/tests/integration/service_test.py
@@ -107,7 +107,7 @@ class ServiceTest(DockerClientTestCase):
         service = self.create_service('db', volumes=['/var/db'])
         container = service.create_container()
         service.start_container(container)
-        self.assertIn('/var/db', container.inspect()['Volumes'])
+        self.assertIn('/var/db', container.get('Volumes'))
 
     def test_create_container_with_cpu_shares(self):
         service = self.create_service('db', cpu_shares=73)
@@ -239,24 +239,27 @@ class ServiceTest(DockerClientTestCase):
             command=['300']
         )
         old_container = service.create_container()
-        self.assertEqual(old_container.dictionary['Config']['Entrypoint'], ['sleep'])
-        self.assertEqual(old_container.dictionary['Config']['Cmd'], ['300'])
-        self.assertIn('FOO=1', old_container.dictionary['Config']['Env'])
+        self.assertEqual(old_container.get('Config.Entrypoint'), ['sleep'])
+        self.assertEqual(old_container.get('Config.Cmd'), ['300'])
+        self.assertIn('FOO=1', old_container.get('Config.Env'))
         self.assertEqual(old_container.name, 'composetest_db_1')
         service.start_container(old_container)
-        volume_path = old_container.inspect()['Volumes']['/etc']
+        old_container.inspect()  # reload volume data
+        volume_path = old_container.get('Volumes')['/etc']
 
         num_containers_before = len(self.client.containers(all=True))
 
         service.options['environment']['FOO'] = '2'
         new_container, = service.recreate_containers()
 
-        self.assertEqual(new_container.dictionary['Config']['Entrypoint'], ['sleep'])
-        self.assertEqual(new_container.dictionary['Config']['Cmd'], ['300'])
-        self.assertIn('FOO=2', new_container.dictionary['Config']['Env'])
+        self.assertEqual(new_container.get('Config.Entrypoint'), ['sleep'])
+        self.assertEqual(new_container.get('Config.Cmd'), ['300'])
+        self.assertIn('FOO=2', new_container.get('Config.Env'))
         self.assertEqual(new_container.name, 'composetest_db_1')
-        self.assertEqual(new_container.inspect()['Volumes']['/etc'], volume_path)
-        self.assertIn(old_container.id, new_container.dictionary['HostConfig']['VolumesFrom'])
+        self.assertEqual(new_container.get('Volumes')['/etc'], volume_path)
+        self.assertIn(
+            'affinity:container==%s' % old_container.id,
+            new_container.get('Config.Env'))
 
         self.assertEqual(len(self.client.containers(all=True)), num_containers_before)
         self.assertNotEqual(old_container.id, new_container.id)
@@ -289,9 +292,7 @@ class ServiceTest(DockerClientTestCase):
         self.assertEqual(old_container.get('Volumes').keys(), ['/data'])
         volume_path = old_container.get('Volumes')['/data']
 
-        service.recreate_containers()
-        new_container = service.containers()[0]
-        service.start_container(new_container)
+        new_container = service.recreate_containers()[0]
         self.assertEqual(new_container.get('Volumes').keys(), ['/data'])
         self.assertEqual(new_container.get('Volumes')['/data'], volume_path)
 
diff --git a/tests/unit/service_test.py b/tests/unit/service_test.py
index 583f72ef0..2ea94edbd 100644
--- a/tests/unit/service_test.py
+++ b/tests/unit/service_test.py
@@ -14,7 +14,9 @@ from compose.service import (
     ConfigError,
     build_port_bindings,
     build_volume_binding,
+    get_container_data_volumes,
     get_container_name,
+    merge_volume_bindings,
     parse_repository_tag,
     parse_volume_spec,
     split_port,
@@ -86,13 +88,6 @@ class ServiceTest(unittest.TestCase):
 
         self.assertEqual(service._get_volumes_from(), [container_id])
 
-    def test_get_volumes_from_previous_container(self):
-        container_id = 'aabbccddee'
-        service = Service('test', image='foo')
-        container = mock.Mock(id=container_id, spec=Container, image='foo')
-
-        self.assertEqual(service._get_volumes_from(container), [container_id])
-
     def test_get_volumes_from_service_container_exists(self):
         container_ids = ['aabbccddee', '12345']
         from_service = mock.create_autospec(Service)
@@ -320,6 +315,9 @@ class ServiceTest(unittest.TestCase):
 
 class ServiceVolumesTest(unittest.TestCase):
 
+    def setUp(self):
+        self.mock_client = mock.create_autospec(docker.Client)
+
     def test_parse_volume_spec_only_one_path(self):
         spec = parse_volume_spec('/the/volume')
         self.assertEqual(spec, (None, '/the/volume', 'rw'))
@@ -345,3 +343,61 @@ class ServiceVolumesTest(unittest.TestCase):
         self.assertEqual(
             binding,
             ('/outside', dict(bind='/inside', ro=False)))
+
+    def test_get_container_data_volumes(self):
+        options = [
+            '/host/volume:/host/volume:ro',
+            '/new/volume',
+            '/existing/volume',
+        ]
+
+        self.mock_client.inspect_image.return_value = {
+            'ContainerConfig': {
+                'Volumes': {
+                    '/mnt/image/data': {},
+                }
+            }
+        }
+        container = Container(self.mock_client, {
+            'Image': 'ababab',
+            'Volumes': {
+                '/host/volume': '/host/volume',
+                '/existing/volume': '/var/lib/docker/aaaaaaaa',
+                '/removed/volume': '/var/lib/docker/bbbbbbbb',
+                '/mnt/image/data': '/var/lib/docker/cccccccc',
+            },
+        }, has_been_inspected=True)
+
+        expected = {
+            '/var/lib/docker/aaaaaaaa': {'bind': '/existing/volume', 'ro': False},
+            '/var/lib/docker/cccccccc': {'bind': '/mnt/image/data', 'ro': False},
+        }
+
+        binds = get_container_data_volumes(container, options)
+        self.assertEqual(binds, expected)
+
+    def test_merge_volume_bindings(self):
+        options = [
+            '/host/volume:/host/volume:ro',
+            '/host/rw/volume:/host/rw/volume',
+            '/new/volume',
+            '/existing/volume',
+        ]
+
+        self.mock_client.inspect_image.return_value = {
+            'ContainerConfig': {'Volumes': {}}
+        }
+
+        intermediate_container = Container(self.mock_client, {
+            'Image': 'ababab',
+            'Volumes': {'/existing/volume': '/var/lib/docker/aaaaaaaa'},
+        }, has_been_inspected=True)
+
+        expected = {
+            '/host/volume': {'bind': '/host/volume', 'ro': True},
+            '/host/rw/volume': {'bind': '/host/rw/volume', 'ro': False},
+            '/var/lib/docker/aaaaaaaa': {'bind': '/existing/volume', 'ro': False},
+        }
+
+        binds = merge_volume_bindings(options, intermediate_container)
+        self.assertEqual(binds, expected)