diff --git a/dvc/cli.py b/dvc/cli.py
index 97d0c0f3f..c25867eba 100644
--- a/dvc/cli.py
+++ b/dvc/cli.py
@@ -14,6 +14,7 @@ from dvc.command.data_sync import CmdDataPush, CmdDataPull, CmdDataFetch
 from dvc.command.status import CmdDataStatus
 from dvc.command.gc import CmdGC
 from dvc.command.add import CmdAdd
+from dvc.command.imp import CmdImport
 from dvc.command.config import CmdConfig
 from dvc.command.checkout import CmdCheckout
 from dvc.command.remote import CmdRemoteAdd, CmdRemoteRemove, CmdRemoteModify, CmdRemoteList
@@ -83,15 +84,28 @@ def parse_args(argv=None):
     init_parser.set_defaults(func=CmdInit)
 
     # Add
-    import_parser = subparsers.add_parser(
+    add_parser = subparsers.add_parser(
                         'add',
                         parents=[parent_parser],
                         help='Add files/directories to dvc')
-    import_parser.add_argument(
+    add_parser.add_argument(
                         'targets',
                         nargs='+',
                         help='Input files/directories')
-    import_parser.set_defaults(func=CmdAdd)
+    add_parser.set_defaults(func=CmdAdd)
+
+    # Import
+    import_parser = subparsers.add_parser(
+                        'import',
+                        parents=[parent_parser],
+                        help='Import files from URL')
+    import_parser.add_argument(
+                        'url',
+                        help='URL')
+    import_parser.add_argument(
+                        'out',
+                        help='Output')
+    import_parser.set_defaults(func=CmdImport)
 
     # Checkout
     checkout_parser = subparsers.add_parser(
diff --git a/dvc/command/imp.py b/dvc/command/imp.py
new file mode 100644
index 000000000..f5119e7be
--- /dev/null
+++ b/dvc/command/imp.py
@@ -0,0 +1,12 @@
+from dvc.exceptions import DvcException
+from dvc.command.common.base import CmdBase
+
+
+class CmdImport(CmdBase):
+    def run(self):
+        try:
+            self.project.imp(self.args.url, self.args.out)
+        except DvcException as ex:
+            self.project.logger.error('Failed to import {}', ex)
+            return 1
+        return 0
diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py
index 525e40209..ba993b21d 100644
--- a/dvc/dependency/base.py
+++ b/dvc/dependency/base.py
@@ -53,3 +53,6 @@ class DependencyBase(object):
 
     def dumpd(self):
         return {self.PARAM_PATH: self.path}
+
+    def download(self, path):
+        self.remote.download(self.path_info, path)
diff --git a/dvc/project.py b/dvc/project.py
index f420d792d..022009751 100644
--- a/dvc/project.py
+++ b/dvc/project.py
@@ -159,6 +159,21 @@ class Project(object):
         stage.dump()
         return stage
 
+    def imp(self, url, out):
+        stage_fname = out + Stage.STAGE_FILE_SUFFIX
+        cwd = os.path.dirname(os.path.abspath(out))
+        stage = Stage.loads(project=self,
+                            cmd=None,
+                            deps=[url],
+                            outs=[out],
+                            fname=stage_fname,
+                            cwd=cwd)
+
+        stage.run()
+        stage.dump()
+        return stage
+
+
     def _reproduce_stage(self, stages, node, force):
         stage = stages[node].reproduce(force=force)
         if not stage:
diff --git a/dvc/remote/base.py b/dvc/remote/base.py
index 721598c1e..9d05ad800 100644
--- a/dvc/remote/base.py
+++ b/dvc/remote/base.py
@@ -26,3 +26,9 @@ class RemoteBase(object):
 
     def checkout(self, path_info, checksum_info):
         pass
+
+    def download(self, path_info, path):
+        pass
+
+    def upload(self, path, path_info):
+        pass
diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py
index 9cad81fed..44e5c7715 100644
--- a/dvc/remote/gs.py
+++ b/dvc/remote/gs.py
@@ -79,8 +79,22 @@ class RemoteGS(RemoteBase):
         if path_info['scheme'] != 'gs':
             raise NotImplementedError
 
-        blob = self.gs.bucket(path_info['key']).get_blob(path_info['key'])
+        blob = self.gs.bucket(path_info['bucket']).get_blob(path_info['key'])
         if not blob:
             return
 
         blob.delete()
+
+    def upload(self, path, path_info):
+        if path_info['scheme'] != 'gs':
+            raise NotImplementedError
+
+        blob = self.gs.bucket(path_info['bucket']).blob(path_info['key'])
+        blob.upload_from_filename(path)
+
+    def download(self, path_info, path):
+        if path_info['scheme'] != 'gs':
+            raise NotImplementedError
+
+        blob = self.gs.bucket(path_info['bucket']).get_blob(path_info['key'])
+        blob.download_to_filename(path)
diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py
index 3f31b41cf..445297747 100644
--- a/dvc/remote/hdfs.py
+++ b/dvc/remote/hdfs.py
@@ -101,3 +101,20 @@ class RemoteHDFS(RemoteBase):
         assert path_info.get('url')
 
         self.rm(path_info['url'])
+
+    def upload(self, path, path_info):
+        if path_info['scheme'] != 'hdfs':
+            raise NotImplementedError
+
+        self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(path_info['url'])))
+        self.hadoop_fs('copyFromLocal {} {}'.format(path, path_info['url']))
+
+    def download(self, path_info, path):
+        if path_info['scheme'] != 'hdfs':
+            raise NotImplementedError
+
+        dname = os.path.dirname(path)
+        if not os.path.exists(dname):
+            os.makedirs(dname)
+
+        self.hadoop_fs('copyToLocal {} {}'.format(path_info['url'], path))
diff --git a/dvc/remote/local.py b/dvc/remote/local.py
index 38901c215..24c00c4ba 100644
--- a/dvc/remote/local.py
+++ b/dvc/remote/local.py
@@ -243,3 +243,22 @@ class RemoteLOCAL(RemoteBase):
             raise NotImplementedError
 
         remove(path_info['path'])
+
+    def _copy(self, from_path, to_path):
+        dname = os.path.dirname(to_path)
+        if not os.path.exists(dname):
+            os.makedirs(dname)
+
+        shutil.copyfile(from_path, to_path)
+
+    def upload(self, path, path_info):
+        if path_info['scheme'] != 'local':
+            raise NotImplementedError
+
+        self._copy(path, path_info['path'])
+
+    def download(self, path_info, path):
+        if path_info['scheme'] != 'local':
+            raise NotImplementedError
+
+        self._copy(path_info['path'], path)
diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py
index 83cce0b15..da5cb951f 100644
--- a/dvc/remote/s3.py
+++ b/dvc/remote/s3.py
@@ -84,3 +84,16 @@ class RemoteS3(RemoteBase):
             obj.delete()
         except Exception:
             pass
+
+    def upload(self, path, path_info):
+        if path_info['scheme'] != 's3':
+            raise NotImplementedError
+
+        self.s3.Object(path_info['bucket'], path_info['key']).upload_file(path)
+
+
+    def download(self, path_info, path):
+        if path_info['scheme'] != 's3':
+            raise NotImplementedError
+
+        self.s3.Object(path_info['bucket'], path_info['key']).download_file(path)
diff --git a/dvc/stage.py b/dvc/stage.py
index 231677923..fb2d9c9f5 100644
--- a/dvc/stage.py
+++ b/dvc/stage.py
@@ -95,6 +95,13 @@ class Stage(object):
     def is_callback(self):
         return not self.is_data_source and len(self.deps) == 0
 
+    @property
+    def is_import(self):
+        return not self.cmd and \
+               len(self.deps) == 1 and \
+               len(self.outs) == 1 and \
+               self.outs[0].path_info['scheme'] == 'local'
+
     def changed(self):
         ret = False
 
@@ -127,12 +134,16 @@ class Stage(object):
         if not self.changed() and not force:
             return None
 
-        if self.cmd:
+        if self.cmd or self.is_import:
             # Removing outputs only if we actually have command to reproduce
             self.remove_outs(ignore_remove=False)
 
+        self.project.logger.info(u'Reproducing \'{}\''.format(self.relpath))
+
         self.run()
 
+        self.project.logger.debug(u'\'{}\' was reproduced'.format(self.relpath))
+
         return self
 
     @staticmethod
@@ -223,8 +234,16 @@ class Stage(object):
             out.save()
 
     def run(self):
-        if not self.is_data_source:
-            self.project.logger.info(u'Reproducing \'{}\':\n\t{}'.format(self.relpath, self.cmd))
+        if self.is_import:
+            msg = u'Importing \'{}\' -> \'{}\''
+            self.project.logger.info(msg.format(self.deps[0].path, self.outs[0].rel_path))
+
+            self.deps[0].download(self.outs[0].path)
+        elif self.is_data_source:
+            self.project.logger.info(u'Verifying data sources in \'{}\''.format(self.relpath))
+            self.check_missing_outputs()
+        else:
+            self.project.logger.info(u'Running command:\n\t{}'.format(self.cmd))
 
             p = subprocess.Popen(self.cmd,
                                  cwd=self.cwd,
@@ -234,14 +253,8 @@ class Stage(object):
             p.communicate()
             if p.returncode != 0:
                raise StageCmdFailedError(self)
- 
-            self.save()
 
-            self.project.logger.debug(u'\'{}\' was reproduced'.format(self.relpath))
-        else:
-            self.project.logger.info(u'Verifying data sources in \'{}\''.format(self.relpath))
-            self.check_missing_outputs()
-            self.save()
+        self.save()
 
     def check_missing_outputs(self):
         missing_outs = [out.rel_path for out in self.outs if not os.path.exists(out.rel_path)]

Implement 'dvc import' for tracking s3/gcp/hdfs files
Even though it can be worked around by running 'dvc run' with awscli/gsutil/hadoop, we should really support this scenario natively, so that we provide a more user-friendly interface and the number of stages in pipelines is reduced for users.
diff --git a/tests/test_repro.py b/tests/test_repro.py
index ff55fc1c0..64129fe22 100644
--- a/tests/test_repro.py
+++ b/tests/test_repro.py
@@ -271,14 +271,22 @@ class TestReproExternalBase(TestDvc):
 
         self.write(self.bucket, foo_key, 'foo')
 
-        stage = self.dvc.run(outs=[out_bar_path],
+        import_stage = self.dvc.imp(out_foo_path, 'import')
+        self.assertTrue(os.path.exists('import'))
+        self.assertTrue(filecmp.cmp('import', self.FOO))
+
+        cmd_stage = self.dvc.run(outs=[out_bar_path],
                              deps=[out_foo_path],
                              cmd='{} {} {}'.format(self.cmd, foo_path, bar_path))
 
         self.write(self.bucket, foo_key, 'bar')
 
-        stages = self.dvc.reproduce(stage.path)
+        stages = self.dvc.reproduce(import_stage.path)
+        self.assertEqual(len(stages), 1)
+        self.assertTrue(os.path.exists('import'))
+        self.assertTrue(filecmp.cmp('import', self.BAR))
 
+        stages = self.dvc.reproduce(cmd_stage.path)
         self.assertEqual(len(stages), 1)
 
 

pip install -e .
parse_log_pytest
requirements.txt
nose>=1.3.7 mock>=2.0.0 coverage codecov xmltodict>=0.11.0 awscli>=1.14.18 pytest
apt-get update apt-get install -y gcc
3.9
requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
altgraph==0.17.4
awscli==1.38.23
bcrypt==4.3.0
binaryornot==0.4.4
boto3==1.7.4
botocore==1.10.84
cachetools==4.2.4
certifi==2025.1.31
cffi==1.17.1
chardet==5.2.0
charset-normalizer==3.4.1
codecov==2.1.13
colorama==0.4.6
configobj==5.0.9
configparser==7.2.0
coverage==7.8.0
cryptography==44.0.2
decorator==5.2.1
dill==0.2.9
docutils==0.16
-e git+https://github.com/iterative/dvc.git@f65ea82ddfecc2da374148d2e282c234681e35d0#egg=dvc
exceptiongroup==1.2.2
future==0.16.0
gapic-google-cloud-datastore-v1==0.15.3
gapic-google-cloud-error-reporting-v1beta1==0.15.3
gapic-google-cloud-logging-v2==0.91.3
gitdb==4.0.12
GitPython==3.1.44
google-api-core==0.1.4
google-auth==1.35.0
google-cloud==0.32.0
google-cloud-bigquery==0.28.0
google-cloud-bigquery-datatransfer==0.1.1
google-cloud-bigtable==0.28.1
google-cloud-container==0.1.1
google-cloud-core==0.28.1
google-cloud-datastore==1.4.0
google-cloud-dns==0.28.0
google-cloud-error-reporting==0.28.0
google-cloud-firestore==0.28.0
google-cloud-language==1.0.2
google-cloud-logging==1.4.0
google-cloud-monitoring==0.28.1
google-cloud-pubsub==0.30.1
google-cloud-resource-manager==0.28.1
google-cloud-runtimeconfig==0.28.1
google-cloud-spanner==0.29.0
google-cloud-speech==0.30.0
google-cloud-storage==1.6.0
google-cloud-trace==0.17.0
google-cloud-translate==1.3.1
google-cloud-videointelligence==1.0.1
google-cloud-vision==0.29.0
google-crc32c==1.7.1
google-gax==0.15.16
google-resumable-media==2.7.2
googleapis-common-protos==1.69.2
grpc-google-iam-v1==0.11.4
grpcio==1.71.0
httplib2==0.22.0
idna==3.10
iniconfig==2.1.0
jmespath==0.10.0
jsonpath-rw==1.4.0
macholib==1.16.3
mock==5.2.0
nanotime==0.5.2
networkx==3.2.1
nose==1.3.7
ntfsutils==0.1.5
oauth2client==3.0.0
packaging==24.2
paramiko==3.5.1
pefile==2024.8.26
pluggy==1.5.0
ply==3.8
proto-google-cloud-datastore-v1==0.90.4
proto-google-cloud-error-reporting-v1beta1==0.15.3
proto-google-cloud-logging-v2==0.91.3
protobuf==3.20.3
psutil==5.9.8
pyasn1==0.6.1
pyasn1_modules==0.4.2
pycparser==2.22
PyInstaller==3.3.1
PyNaCl==1.5.0
pyparsing==3.2.3
pytest==8.3.5
python-dateutil==2.9.0.post0
pytz==2025.2
PyYAML==6.0.2
reflink==0.2.0
requests==2.32.3
rsa==4.7.2
s3transfer==0.1.13
schema==0.7.7
six==1.17.0
smmap==5.0.2
tomli==2.2.1
urllib3==1.26.20
xmltodict==0.14.2
zc.lockfile==3.0.post1

name: dvc
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=3.0.16=h5eee18b_0
  - pip=25.0=py39h06a4308_0
  - python=3.9.21=he870216_1
  - readline=8.2=h5eee18b_0
  - setuptools=75.8.0=py39h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - tzdata=2025a=h04d1e81_0
  - wheel=0.45.1=py39h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - altgraph==0.17.4
      - awscli==1.38.23
      - bcrypt==4.3.0
      - binaryornot==0.4.4
      - boto3==1.7.4
      - botocore==1.10.84
      - cachetools==4.2.4
      - certifi==2025.1.31
      - cffi==1.17.1
      - chardet==5.2.0
      - charset-normalizer==3.4.1
      - codecov==2.1.13
      - colorama==0.4.6
      - configobj==5.0.9
      - configparser==7.2.0
      - coverage==7.8.0
      - cryptography==44.0.2
      - decorator==5.2.1
      - dill==0.2.9
      - docutils==0.16
      - exceptiongroup==1.2.2
      - future==0.16.0
      - gapic-google-cloud-datastore-v1==0.15.3
      - gapic-google-cloud-error-reporting-v1beta1==0.15.3
      - gapic-google-cloud-logging-v2==0.91.3
      - gitdb==4.0.12
      - gitpython==3.1.44
      - google-api-core==0.1.4
      - google-auth==1.35.0
      - google-cloud==0.32.0
      - google-cloud-bigquery==0.28.0
      - google-cloud-bigquery-datatransfer==0.1.1
      - google-cloud-bigtable==0.28.1
      - google-cloud-container==0.1.1
      - google-cloud-core==0.28.1
      - google-cloud-datastore==1.4.0
      - google-cloud-dns==0.28.0
      - google-cloud-error-reporting==0.28.0
      - google-cloud-firestore==0.28.0
      - google-cloud-language==1.0.2
      - google-cloud-logging==1.4.0
      - google-cloud-monitoring==0.28.1
      - google-cloud-pubsub==0.30.1
      - google-cloud-resource-manager==0.28.1
      - google-cloud-runtimeconfig==0.28.1
      - google-cloud-spanner==0.29.0
      - google-cloud-speech==0.30.0
      - google-cloud-storage==1.6.0
      - google-cloud-trace==0.17.0
      - google-cloud-translate==1.3.1
      - google-cloud-videointelligence==1.0.1
      - google-cloud-vision==0.29.0
      - google-crc32c==1.7.1
      - google-gax==0.15.16
      - google-resumable-media==2.7.2
      - googleapis-common-protos==1.69.2
      - grpc-google-iam-v1==0.11.4
      - grpcio==1.71.0
      - httplib2==0.22.0
      - idna==3.10
      - iniconfig==2.1.0
      - jmespath==0.10.0
      - jsonpath-rw==1.4.0
      - macholib==1.16.3
      - mock==5.2.0
      - nanotime==0.5.2
      - networkx==3.2.1
      - nose==1.3.7
      - ntfsutils==0.1.5
      - oauth2client==3.0.0
      - packaging==24.2
      - paramiko==3.5.1
      - pefile==2024.8.26
      - pluggy==1.5.0
      - ply==3.8
      - proto-google-cloud-datastore-v1==0.90.4
      - proto-google-cloud-error-reporting-v1beta1==0.15.3
      - proto-google-cloud-logging-v2==0.91.3
      - protobuf==3.20.3
      - psutil==5.9.8
      - pyasn1==0.6.1
      - pyasn1-modules==0.4.2
      - pycparser==2.22
      - pyinstaller==3.3.1
      - pynacl==1.5.0
      - pyparsing==3.2.3
      - pytest==8.3.5
      - python-dateutil==2.9.0.post0
      - pytz==2025.2
      - pyyaml==6.0.2
      - reflink==0.2.0
      - requests==2.32.3
      - rsa==4.7.2
      - s3transfer==0.1.13
      - schema==0.7.7
      - six==1.17.0
      - smmap==5.0.2
      - tomli==2.2.1
      - urllib3==1.26.20
      - xmltodict==0.14.2
      - zc-lockfile==3.0.post1
prefix: /opt/conda/envs/dvc

tests/test_repro.py::TestReproExternalLOCAL::test
tests/test_repro.py::TestReproMissingMd5InStageFile::test tests/test_repro.py::TestReproShell::test
tests/test_repro.py::TestReproNoDeps::test tests/test_repro.py::TestReproForce::test tests/test_repro.py::TestReproChangedCode::test tests/test_repro.py::TestReproChangedData::test tests/test_repro.py::TestReproChangedDeepData::test tests/test_repro.py::TestReproPhony::test tests/test_repro.py::TestNonExistingOutput::test tests/test_repro.py::TestReproDataSource::test tests/test_repro.py::TestReproChangedDir::test tests/test_repro.py::TestCmdRepro::test tests/test_repro.py::TestCmdReproChdir::test tests/test_repro.py::TestReproExternalBase::test tests/test_repro.py::TestReproExternalS3::test tests/test_repro.py::TestReproExternalGS::test tests/test_repro.py::TestReproExternalHDFS::test
dvc/stage.py dvc/remote/s3.py dvc/remote/gs.py dvc/project.py dvc/cli.py dvc/remote/base.py dvc/remote/local.py dvc/remote/hdfs.py dvc/dependency/base.py dvc/command/imp.py
dvc/stage.py dvc/remote/s3.py dvc/remote/gs.py dvc/project.py dvc/cli.py dvc/remote/base.py dvc/remote/local.py dvc/remote/hdfs.py dvc/dependency/base.py dvc/command/imp.py