diff --git a/airflow/contrib/sensors/sftp_sensor.py b/airflow/contrib/sensors/sftp_sensor.py
index 0b61c801514cc1..23df400cb5e4e1 100644
--- a/airflow/contrib/sensors/sftp_sensor.py
+++ b/airflow/contrib/sensors/sftp_sensor.py
@@ -38,9 +38,11 @@ class SFTPSensor(BaseSensorOperator):
     def __init__(self, path, sftp_conn_id='sftp_default', *args, **kwargs):
         super(SFTPSensor, self).__init__(*args, **kwargs)
         self.path = path
-        self.hook = SFTPHook(sftp_conn_id)
+        self.hook = None
+        self.sftp_conn_id = sftp_conn_id
 
     def poke(self, context):
+        self.hook = SFTPHook(self.sftp_conn_id)
         self.log.info('Poking for %s', self.path)
         try:
             self.hook.get_mod_time(self.path)
diff --git a/tests/contrib/sensors/test_sftp_sensor.py b/tests/contrib/sensors/test_sftp_sensor.py
index 0fb9d6c610ca9d..e34faeb944f1b5 100644
--- a/tests/contrib/sensors/test_sftp_sensor.py
+++ b/tests/contrib/sensors/test_sftp_sensor.py
@@ -67,3 +67,9 @@ def test_sftp_failure(self, sftp_hook_mock):
             sftp_sensor.poke(context)
             sftp_hook_mock.return_value.get_mod_time.assert_called_with(
                 '/path/to/file/1970-01-01.txt')
+
+    def test_hook_not_created_during_init(self):
+        sftp_sensor = SFTPSensor(
+            task_id='unit_test',
+            path='/path/to/file/1970-01-01.txt')
+        self.assertIsNone(sftp_sensor.hook)