diff --git a/Makefile b/Makefile
index ac580d2504..5d8af99659 100644
--- a/Makefile
+++ b/Makefile
@@ -36,7 +36,7 @@ endif
 requirements:
 	pip install -U -r requirements/pre.txt
 	pip install -U -r requirements/default.txt --no-cache-dir --upgrade-strategy only-if-needed
-	pip install -U -r requirements/extra.txt --no-cache-dir
+	pip install -U -r requirements/extra.txt --no-cache-dir --upgrade-strategy only-if-needed
 
 test-requirements: requirements
 	pip install -U -r requirements/test.txt --no-cache-dir --upgrade-strategy only-if-needed
@@ -89,7 +89,7 @@ coverage: test coverage-local
 docs-requirements:
 	pip install -U -r requirements/pre.txt
 	pip install -U -r requirements/docs.txt --no-cache-dir --upgrade-strategy only-if-needed
-	pip install -U -r requirements/extra.txt --no-cache-dir
+	pip install -U -r requirements/extra.txt --no-cache-dir --upgrade-strategy only-if-needed
 	python setup.py install --force
 
 docs-local:
diff --git a/config/devstack.cfg b/config/devstack.cfg
index 11fdabda85..3a3c6ee00c 100644
--- a/config/devstack.cfg
+++ b/config/devstack.cfg
@@ -23,9 +23,9 @@ engine = hadoop
 marker = hdfs://localhost:9000/edx-analytics-pipeline/marker/
 
 [event-logs]
-pattern = .*tracking.log.*
+pattern = [".*tracking.log.*"]
 expand_interval = 2 days
-source = hdfs://localhost:9000/data/
+source = ["hdfs://localhost:9000/data/"]
 
 [event-export]
 output_root = hdfs://localhost:9000/edx-analytics-pipeline/event-export/output/
@@ -58,13 +58,6 @@ output_root = hdfs://localhost:9000/edx-analytics-pipeline/activity/
 [enrollments]
 interval_start = 2013-11-01
 
-[enrollment-reports]
-src = hdfs://localhost:9000/data/
-destination = hdfs://localhost:9000/edx-analytics-pipeline/enrollment_reports/output/
-offsets = hdfs://localhost:9000/edx-analytics-pipeline/enrollment_reports/offsets.tsv
-blacklist = hdfs://localhost:9000/edx-analytics-pipeline/enrollment_reports/course_blacklist.tsv
-history = hdfs://localhost:9000/edx-analytics-pipeline/enrollment_reports/enrollment_history.tsv
-
 [financial-reports]
 shoppingcart-partners = {"DEFAULT": "edx"}
 
@@ -79,7 +72,7 @@ dropoff_threshold = 0.05
 
 [elasticsearch]
 # Point to the vagrant host's port 9201 where we assume elasticsearch is running
-host = http://172.17.0.1:9201/
+host = ["http://172.17.0.1:9201/"]
 
 [module-engagement]
 alias = roster
diff --git a/config/test.cfg b/config/test.cfg
index 2d31b7e4fd..c83e15ba88 100644
--- a/config/test.cfg
+++ b/config/test.cfg
@@ -21,17 +21,14 @@ marker = s3://fake/marker/
 remote_log_level = DEBUG
 
 [event-logs]
-pattern = .*tracking.log-(?P<date>\d{8}).*\.gz
-          .*tracking.notalog-(?P<date>\d{8}).*\.gz
+# pattern = [".*tracking.log-(?P<date>\d{8}).*\.gz", ".*tracking.notalog-(?P<date>\d{8}).*\.gz"]
+pattern = [".*tracking.log-(?P<date>\\\\d{8}).*\\\\.gz", ".*tracking.notalog-(?P<date>\\\\d{8}).*\\\\.gz"]
 expand_interval = 2 days
-source = s3://fake/input/
-         s3://fake/input2/
+source = ["s3://fake/input/", "s3://fake/input2/"]
 
 [segment-logs]
-pattern = .*segment.log-(?P<date>\d{8}).*\.gz
-          .*segment.notalog-(?P<date>\d{8}).*\.gz
-source = s3://fake/segment/input/
-         s3://fake/segment/input2/
+pattern = [".*segment.log-(?P<date>\\\\d{8}).*\\\\.gz", ".*segment.notalog-(?P<date>\\\\d{8}).*\\\\.gz"]
+source = ["s3://fake/segment/input/", "s3://fake/segment/input2/"]
 
 [event-export]
 output_root = s3://fake/
@@ -69,13 +66,6 @@ overwrite_n_days = 14
 interval_start = 2013-11-01
 overwrite_n_days = 14
 
-[enrollment-reports]
-src = s3://fake/input/
-destination = s3://fake/enrollment_reports/output/
-offsets = s3://fake/enrollment_reports/offsets.tsv
-blacklist = s3://fake/enrollment_reports/course_blacklist.tsv
-history = s3://fake/enrollment_reports/enrollment_history.tsv
-
 [financial-reports]
 shoppingcart-partners = {"OpenCraftX": "OCX", "DEFAULT": "edx"}
 
@@ -96,9 +86,9 @@ access_token = acceptance
 credentials = s3://fake/vertica_creds.json
 schema = testing
 read_timeout = 600
-standard_roles = data_engineering_team
-restricted_roles = data_engineering_team
-business_intelligence_team_roles = data_engineering_team
+standard_roles = ["data_engineering_team"]
+restricted_roles = ["data_engineering_team"]
+business_intelligence_team_roles = ["data_engineering_team"]
 event_retention_interval = 3 days
 
 [payment-reconciliation]
@@ -129,8 +119,7 @@ host=test.cybersource.com/test
 interval_start = 2015-09-01
 
 [payment]
-cybersource_merchant_ids = test
-                           empty_test
+cybersource_merchant_ids = ["test", "empty_test"]
 
 [videos]
 dropoff_threshold = 0.05
@@ -141,7 +130,7 @@ alias = roster
 number_of_shards = 5
 
 [course-catalog-api]
-partner_short_codes = openedx
+partner_short_codes = ["openedx"]
 api_root_url = http://example.com/api/v1/
 
 [ccx]
diff --git a/docs/source/gen_tasks.py b/docs/source/gen_tasks.py
index e95b0a3b36..31260a80ed 100644
--- a/docs/source/gen_tasks.py
+++ b/docs/source/gen_tasks.py
@@ -7,7 +7,7 @@
 import sys
 import time
 import stevedore
-from luigi.task import Register
+from luigi.task_register import Register
 
 
 SPHINX_DIR = os.path.dirname(__file__)
@@ -81,7 +81,7 @@ def gen_sphinx_tasks(entry_point, labels, *_args, **kwargs):
         sys.exit('Unable to write to {file_name}'.format(file_name=tocfile_name))
 
     # For each Task, sorted by class name
-    tasks = Register.get_reg()
+    tasks = Register._get_reg()
     for name in sorted(tasks):
         cls = tasks[name]
         module = cls.__module__
diff --git a/docs/source/luigi_sphinx.py b/docs/source/luigi_sphinx.py
index 742af7a033..8446d18a6f 100644
--- a/docs/source/luigi_sphinx.py
+++ b/docs/source/luigi_sphinx.py
@@ -5,6 +5,7 @@
 import inspect
 import luigi
 
+from luigi.parameter import _no_value
 
 def append_parameters(_app, _what, _name, obj, _options, lines):
     """
@@ -36,13 +37,13 @@ def append_parameters(_app, _what, _name, obj, _options, lines):
 
                 # Mark configured parameters (requires protected-access)
                 # pylint: disable=W0212
-                if hasattr(membervalue, '_Parameter__config') and membervalue._Parameter__config is not None:
-                    param['default'] = ' pulled from ``{section}.{name}``'.format(**membervalue._Parameter__config)
+                if hasattr(membervalue, '_config_path') and membervalue._config_path is not None:
+                    param['default'] = 'pulled from ``{section}.{name}``'.format(**membervalue._config_path)
                     param['type'] = u'{type}, configurable'.format(**param)
 
                 # Mark optional parameters
-                elif hasattr(membervalue, 'default'):
-                    param['default'] = membervalue.default
+                elif hasattr(membervalue, '_default') and membervalue._default != _no_value:
+                    param['default'] = membervalue._default
                     param['type'] = u'{type}, optional'.format(**param)
 
                 if 'default' in param:
diff --git a/edx/analytics/tasks/common/elasticsearch_load.py b/edx/analytics/tasks/common/elasticsearch_load.py
index 384cf2aff6..82a03152cd 100644
--- a/edx/analytics/tasks/common/elasticsearch_load.py
+++ b/edx/analytics/tasks/common/elasticsearch_load.py
@@ -47,8 +47,7 @@ class ElasticsearchIndexTask(OverwriteOutputMixin, MapReduceJobTask):
 
     """
 
-    host = luigi.Parameter(
-        is_list=True,
+    host = luigi.ListParameter(
         config_path={'section': 'elasticsearch', 'name': 'host'},
         description='Hostnames for the elasticsearch cluster nodes. They can be specified in any of the formats'
                     ' accepted by the elasticsearch-py library. This includes complete URLs such as http://foo.com/, or'
diff --git a/edx/analytics/tasks/common/mapreduce.py b/edx/analytics/tasks/common/mapreduce.py
index 0b894613b9..9b2628c049 100644
--- a/edx/analytics/tasks/common/mapreduce.py
+++ b/edx/analytics/tasks/common/mapreduce.py
@@ -11,8 +11,7 @@
 from hashlib import md5
 
 import luigi
-import luigi.hadoop
-import luigi.hdfs
+import luigi.contrib.hadoop
 import luigi.task
 from luigi import configuration
 
@@ -37,8 +36,7 @@ class MapReduceJobTaskMixin(object):
         description='The input_format for Hadoop job to use. For example, when '
         'running with manifest file, specify "oddjob.ManifestTextInputFormat" for input_format.',
     )
-    lib_jar = luigi.Parameter(
-        is_list=True,
+    lib_jar = luigi.ListParameter(
         default=[],
         significant=False,
         description='A list of library jars that the Hadoop job can make use of.',
@@ -55,7 +53,7 @@ class MapReduceJobTaskMixin(object):
     )
 
 
-class MapReduceJobTask(MapReduceJobTaskMixin, luigi.hadoop.JobTask):
+class MapReduceJobTask(MapReduceJobTaskMixin, luigi.contrib.hadoop.JobTask):
     """
     Execute a map reduce job.  Typically using Hadoop, but can execute the
     job in process as well.
@@ -133,7 +131,7 @@ def input_hadoop(self):
         return convert_to_manifest_input_if_necessary(manifest_id, super(MapReduceJobTask, self).input_hadoop())
 
 
-class MapReduceJobRunner(luigi.hadoop.HadoopJobRunner):
+class MapReduceJobRunner(luigi.contrib.hadoop.HadoopJobRunner):
     """
     Support more customization of the streaming command.
 
@@ -160,11 +158,11 @@ def __init__(self, libjars_in_hdfs=None, input_format=None):
         )
 
 
-class EmulatedMapReduceJobRunner(luigi.hadoop.JobRunner):
+class EmulatedMapReduceJobRunner(luigi.contrib.hadoop.JobRunner):
     """
     Execute map reduce tasks in process on the machine that is running luigi.
 
-    This is a modified version of luigi.hadoop.LocalJobRunner. The key differences are:
+    This is a modified version of luigi.contrib.hadoop.LocalJobRunner. The key differences are:
 
     * It gracefully handles .gz input files, decompressing them and streaming them directly to the mapper. This mirrors
       the behavior of hadoop's default file input format. Note this only works for files that support `tell()` and
@@ -252,7 +250,7 @@ class MultiOutputMapReduceJobTask(MapReduceJobTask):
     output_root = luigi.Parameter(
         description='A URL location where the split files will be stored.',
     )
-    delete_output_root = luigi.BooleanParameter(
+    delete_output_root = luigi.BoolParameter(
         default=False,
         significant=False,
         description='If True, recursively deletes the `output_root` at task creation.',
diff --git a/edx/analytics/tasks/common/mysql_load.py b/edx/analytics/tasks/common/mysql_load.py
index b8da2884dc..da2d7c8082 100644
--- a/edx/analytics/tasks/common/mysql_load.py
+++ b/edx/analytics/tasks/common/mysql_load.py
@@ -59,8 +59,10 @@ def requires(self):
         if self.required_tasks is None:
             self.required_tasks = {
                 'credentials': ExternalURL(url=self.credentials),
-                'insert_source': self.insert_source_task
             }
+            if not self.insert_source_task_dynamically:
+                self.required_tasks['insert_source'] = self.insert_source_task
+
         return self.required_tasks
 
     @property
@@ -68,6 +70,11 @@ def insert_source_task(self):
         """Defines task that provides source of data for insertion."""
         raise NotImplementedError
 
+    @property
+    def insert_source_task_dynamically(self):
+        """Declare if task that provides source of data for insertion should be a dynamic dependency."""
+        return False
+
     @property
     def table(self):
         """Provides name of database table."""
@@ -170,9 +177,11 @@ def create_database(self):
     def rows(self):
         """Return/yield tuples or lists corresponding to each row to be inserted """
         try:
-            with self.input()['insert_source'].open('r') as fobj:
-                for line in fobj:
-                    yield line.strip('\n').split('\t')
+            if self.insert_source_task is not None:
+                input_target = self.insert_source_task.output()
+                with input_target.open('r') as fobj:
+                    for line in fobj:
+                        yield line.strip('\n').split('\t')
         except RuntimeError:
             # While calling finish on an input target, Luigi throws a RuntimeError exception if the subprocess command
             # to read the input returns a non-zero return code. As all of the data's been read already, we choose to ignore
@@ -320,6 +329,14 @@ def run(self):
 
         Normally you don't want to override this.
         """
+        # Use dynamic dependencies here to make sure that the tasks on
+        # which this depends have been run.
+        if self.insert_source_task_dynamically and self.insert_source_task is not None:
+            log.debug('Yielding dependency dynamically at runtime for %s: %s', self, self.insert_source_task)
+            yield self.insert_source_task
+        else:
+            yield []
+
         if not (self.table and self.columns):
             raise Exception("table and columns need to be specified")
 
diff --git a/edx/analytics/tasks/common/pathutil.py b/edx/analytics/tasks/common/pathutil.py
index f165fa76df..ee6cacd6eb 100644
--- a/edx/analytics/tasks/common/pathutil.py
+++ b/edx/analytics/tasks/common/pathutil.py
@@ -13,10 +13,10 @@
 import re
 
 import luigi
-import luigi.format
-import luigi.hdfs
+import luigi.contrib.hdfs
+import luigi.contrib.hdfs.format
 import luigi.task
-from luigi.date_interval import DateInterval
+from luigi.date_interval import Custom
 
 from edx.analytics.tasks.util import eventlog
 from edx.analytics.tasks.util.s3_util import ScalableS3Client, generate_s3_sources, get_s3_bucket_key_names
@@ -30,13 +30,11 @@ class PathSetTask(luigi.Task):
     A task to select a subset of files in an S3 bucket or local FS.
 
     """
-    src = luigi.Parameter(
-        is_list=True,
+    src = luigi.ListParameter(
         config_path={'section': 'event-logs', 'name': 'source'},
         description='A URL pointing to a folder in s3:// or local FS.',
     )
-    include = luigi.Parameter(
-        is_list=True,
+    include = luigi.ListParameter(
         default=('*',),
         description='A list of patterns to use to select.  Multiple patterns are OR\'d.',
     )
@@ -44,7 +42,7 @@ class PathSetTask(luigi.Task):
         default=None,
         description='A URL pointing to a manifest file location.',
     )
-    include_zero_length = luigi.BooleanParameter(
+    include_zero_length = luigi.BoolParameter(
         default=False,
         description='If True, include files/directories with size zero.',
     )
@@ -64,7 +62,7 @@ def generate_file_list(self):
                     source = url_path_join(src, path)
                     yield ExternalURL(source)
             elif src.startswith('hdfs'):
-                for source, size in luigi.hdfs.listdir(src, recursive=True, include_size=True):
+                for source, size in luigi.contrib.hdfs.listdir(src, recursive=True, include_size=True):
                     if not self.include_zero_length and size == 0:
                         continue
                     elif any(fnmatch.fnmatch(source, include_val) for include_val in self.include):
@@ -109,8 +107,7 @@ def output(self):
 class EventLogSelectionDownstreamMixin(object):
     """Defines parameters for passing upstream to tasks that use EventLogSelectionMixin."""
 
-    source = luigi.Parameter(
-        is_list=True,
+    source = luigi.ListParameter(
         config_path={'section': 'event-logs', 'name': 'source'},
         description='A URL to a path that contains log files that contain the events. (e.g., s3://my_bucket/foo/).',
     )
@@ -122,8 +119,7 @@ class EventLogSelectionDownstreamMixin(object):
         description='A time interval to add to the beginning and end of the interval to expand the windows of '
         'files captured.',
     )
-    pattern = luigi.Parameter(
-        is_list=True,
+    pattern = luigi.ListParameter(
         config_path={'section': 'event-logs', 'name': 'pattern'},
         description='A regex with a named capture group for the date that approximates the date that the events '
         'within were emitted. Note that the search interval is expanded, so events don\'t have to be in exactly '
@@ -150,7 +146,7 @@ class PathSelectionByDateIntervalTask(EventLogSelectionDownstreamMixin, luigi.Wr
 
     def __init__(self, *args, **kwargs):
         super(PathSelectionByDateIntervalTask, self).__init__(*args, **kwargs)
-        self.interval = DateInterval(
+        self.interval = Custom(
             self.interval.date_a - self.expand_interval,
             self.interval.date_b + self.expand_interval
         )
@@ -200,9 +196,9 @@ def _get_s3_urls(self, source):
 
     def _get_hdfs_urls(self, source):
         """Recursively list all files inside the source directory on the hdfs filesystem."""
-        if luigi.hdfs.exists(source):
+        if luigi.contrib.hdfs.exists(source):
             # listdir raises an exception if the source doesn't exist.
-            for source in luigi.hdfs.listdir(source, recursive=True):
+            for source in luigi.contrib.hdfs.listdir(source, recursive=True):
                 yield source
 
     def _get_local_urls(self, source):
diff --git a/edx/analytics/tasks/common/sqoop.py b/edx/analytics/tasks/common/sqoop.py
index 9825b939ef..68ba76523e 100644
--- a/edx/analytics/tasks/common/sqoop.py
+++ b/edx/analytics/tasks/common/sqoop.py
@@ -7,8 +7,8 @@
 
 import luigi
 import luigi.configuration
-import luigi.hadoop
-import luigi.hdfs
+import luigi.contrib.hadoop
+import luigi.contrib.hdfs
 
 from edx.analytics.tasks.util.overwrite import OverwriteOutputMixin
 from edx.analytics.tasks.util.url import ExternalURL, get_target_from_url, url_path_join
@@ -57,7 +57,7 @@ class SqoopImportMixin(object):
         significant=False,
         description='The number of map tasks to ask Sqoop to use.',
     )
-    verbose = luigi.BooleanParameter(
+    verbose = luigi.BoolParameter(
         default=False,
         significant=False,
         description='Print more information while working.',
@@ -70,7 +70,7 @@ class SqoopImportMixin(object):
     )
 
 
-class SqoopImportTask(OverwriteOutputMixin, SqoopImportMixin, luigi.hadoop.BaseHadoopJobTask):
+class SqoopImportTask(OverwriteOutputMixin, SqoopImportMixin, luigi.contrib.hadoop.BaseHadoopJobTask):
     """
     An abstract task that uses Sqoop to read data out of a database and
     writes it to a file in CSV format.
@@ -82,8 +82,7 @@ class SqoopImportTask(OverwriteOutputMixin, SqoopImportMixin, luigi.hadoop.BaseH
     table_name = luigi.Parameter(
         description='The name of the table to import.',
     )
-    columns = luigi.Parameter(
-        is_list=True,
+    columns = luigi.ListParameter(
         default=[],
         description='A list of column names to be included.  Default is to include all columns.'
     )
@@ -209,11 +208,11 @@ class SqoopImportFromMysql(SqoopImportTask):
     * delimiters optionally enclosed by single quotes (')
 
     """
-    mysql_delimiters = luigi.BooleanParameter(
+    mysql_delimiters = luigi.BoolParameter(
         default=True,
         description='Use standard mysql delimiters (on by default).',
     )
-    direct = luigi.BooleanParameter(
+    direct = luigi.BoolParameter(
         default=True,
         significant=False,
         description='Use mysqldumpi\'s "direct" mode.  Requires that no set of columns be selected.',
@@ -233,13 +232,13 @@ def import_args(self):
         return arglist
 
 
-class SqoopPasswordTarget(luigi.hdfs.HdfsTarget):
+class SqoopPasswordTarget(luigi.contrib.hdfs.HdfsTarget):
     """Defines a temp file in HDFS to hold password."""
     def __init__(self):
         super(SqoopPasswordTarget, self).__init__(is_tmp=True)
 
 
-class SqoopImportRunner(luigi.hadoop.JobRunner):
+class SqoopImportRunner(luigi.contrib.hadoop.JobRunner):
     """Runs a SqoopImportTask by shelling out to sqoop."""
 
     def run_job(self, job):
@@ -274,7 +273,7 @@ def run_job(self, job):
             # (using __del__()), but safer to just make sure.
             password_target = SqoopPasswordTarget()
             arglist = job.get_arglist(password_target)
-            luigi.hadoop.run_and_track_hadoop_job(arglist)
+            luigi.contrib.hadoop.run_and_track_hadoop_job(arglist)
         finally:
             password_target.remove()
             metadata['end_time'] = datetime.datetime.utcnow().isoformat()
diff --git a/edx/analytics/tasks/common/tests/map_reduce_mixins.py b/edx/analytics/tasks/common/tests/map_reduce_mixins.py
index 2f3347d7ee..b1e34cbbd4 100644
--- a/edx/analytics/tasks/common/tests/map_reduce_mixins.py
+++ b/edx/analytics/tasks/common/tests/map_reduce_mixins.py
@@ -28,7 +28,7 @@ class MapperTestMixin(object):
         'mapreduce_engine': 'local',
         'user_country_output': 'test://output/',
         'name': 'test',
-        'src': ['test://input/'],
+        'src': '["test://input/"]',
         'dest': 'test://output/',
         'overwrite_from_date': datetime.date(2014, 4, 1),
     }
@@ -147,7 +147,7 @@ class ReducerTestMixin(object):
         'mapreduce_engine': 'local',
         'user_country_output': 'test://output/',
         'name': 'test',
-        'src': ['test://input/'],
+        'src': '["test://input/"]',
         'dest': 'test://output/',
         'date': datetime.datetime.strptime('2014-04-01', '%Y-%m-%d').date(),
         'overwrite_from_date': datetime.date(2014, 4, 1),
diff --git a/edx/analytics/tasks/common/tests/test_elasticsearch_load.py b/edx/analytics/tasks/common/tests/test_elasticsearch_load.py
index f70ac15fc5..34b0860f9e 100644
--- a/edx/analytics/tasks/common/tests/test_elasticsearch_load.py
+++ b/edx/analytics/tasks/common/tests/test_elasticsearch_load.py
@@ -4,7 +4,7 @@
 import unittest
 
 import ddt
-import luigi.hdfs
+import luigi.contrib.hdfs.target
 from elasticsearch import TransportError
 from freezegun import freeze_time
 from mock import call, patch
@@ -338,7 +338,7 @@ def test_indexing_failures(self):
 
 
 @freeze_time('2016-03-25')
-@patch.object(luigi.hdfs.HdfsTarget, '__del__', return_value=None)
+@patch.object(luigi.contrib.hdfs.target.HdfsTarget, '__del__', return_value=None)
 class ElasticsearchIndexTaskCommitTest(BaseIndexTest, ReducerTestMixin, unittest.TestCase):
     """Tests for the commit logic."""
 
diff --git a/edx/analytics/tasks/common/tests/test_mapreduce.py b/edx/analytics/tasks/common/tests/test_mapreduce.py
index 6aba50afc8..6eb0a591a6 100644
--- a/edx/analytics/tasks/common/tests/test_mapreduce.py
+++ b/edx/analytics/tasks/common/tests/test_mapreduce.py
@@ -2,13 +2,14 @@
 
 from __future__ import absolute_import
 
+import json
 import os
 import shutil
 import tempfile
 import unittest
 
 import luigi
-import luigi.hdfs
+from luigi.contrib.hdfs.target import HdfsTarget
 from mock import call, patch
 
 from edx.analytics.tasks.common.mapreduce import MapReduceJobTask, MultiOutputMapReduceJobTask
@@ -67,11 +68,11 @@ def test_missing_input_format(self):
 class TaskWithSpecialOutputs(luigi.ExternalTask):
     """A task with a single output that requires the use of a configurable library jar and input format."""
 
-    lib_jar_path = luigi.Parameter(default=[], is_list=True)
+    lib_jar_path = luigi.ListParameter(default=[])
     input_format = luigi.Parameter(default=None)
 
     def output(self):
-        target = luigi.hdfs.HdfsTarget('/tmp/foo')
+        target = HdfsTarget('/tmp/foo')
         target.lib_jar = self.lib_jar_path
         target.input_format = self.input_format
         return target
@@ -121,8 +122,7 @@ class MultiOutputMapReduceJobTaskOutputRootTest(unittest.TestCase):
     """Tests for output_root behavior of MultiOutputMapReduceJobTask."""
 
     def setUp(self):
-        # Define a real output directory, so it can
-        # be removed if existing.
+        # Define a real output directory, so it can be removed if existing.
         def cleanup(dirname):
             """Remove the temp directory only if it exists."""
             if os.path.exists(dirname):
@@ -131,13 +131,10 @@ def cleanup(dirname):
         self.output_root = tempfile.mkdtemp()
         self.addCleanup(cleanup, self.output_root)
 
-        patcher = patch('edx.analytics.tasks.common.mapreduce.luigi.configuration.get_config')
-        self.mock_get_config = patcher.start()
-        self.addCleanup(patcher.stop)
-
     def test_no_delete_output_root(self):
         self.assertTrue(os.path.exists(self.output_root))
         TestJobTask(
+            delete_output_root=False,
             mapreduce_engine='local',
             output_root=self.output_root,
         )
@@ -145,24 +142,27 @@ def test_no_delete_output_root(self):
 
     def test_delete_output_root(self):
         temporary_file_path = tempfile.mkdtemp()
-        self.mock_get_config.return_value.get.return_value = temporary_file_path
         self.addCleanup(shutil.rmtree, temporary_file_path)
 
         # We create a task in order to get the output path.
         task = TestJobTask(
             mapreduce_engine='local',
             output_root=self.output_root,
+            delete_output_root=False,
+            marker=temporary_file_path,
         )
         output_marker = task.output().path
         open(output_marker, 'a').close()
+        self.assertTrue(os.path.exists(self.output_root))
         self.assertTrue(task.complete())
 
         # Once the output path is created, we can
-        # then confirm that it gets cleaned up..
+        # then confirm that it gets cleaned up.
         task = TestJobTask(
             mapreduce_engine='local',
             output_root=self.output_root,
-            delete_output_root="true",
+            delete_output_root=True,
+            marker=temporary_file_path,
         )
         self.assertFalse(task.complete())
         self.assertFalse(os.path.exists(self.output_root))
diff --git a/edx/analytics/tasks/common/tests/test_mysql_load.py b/edx/analytics/tasks/common/tests/test_mysql_load.py
index 0c084e1a56..f6dbdf7ba8 100644
--- a/edx/analytics/tasks/common/tests/test_mysql_load.py
+++ b/edx/analytics/tasks/common/tests/test_mysql_load.py
@@ -8,7 +8,7 @@
 
 import luigi
 import luigi.task
-from mock import MagicMock, call, patch, sentinel
+from mock import MagicMock, PropertyMock, call, patch, sentinel
 
 from edx.analytics.tasks.common.mysql_load import MysqlInsertTask, coerce_for_mysql_connect
 from edx.analytics.tasks.util.tests.config import with_luigi_config
@@ -94,18 +94,20 @@ def create_task(self, credentials=None, source=None, insert_chunk_size=100, over
 
         fake_input = {
             'credentials': FakeTarget(value=textwrap.dedent(credentials)),
-            'insert_source': FakeTarget(value=textwrap.dedent(source))
         }
-
         task.input = MagicMock(return_value=fake_input)
+
+        cls.insert_source_task = PropertyMock(return_value=MagicMock())
+        task.insert_source_task.output = MagicMock(return_value=FakeTarget(value=textwrap.dedent(source)))
+
         return task
 
     def test_connect_with_credential_syntax_error(self):
         with self.assertRaises(ValueError):
-            self.create_task(credentials='{').run()
+            list(self.create_task(credentials='{').run())
 
     def test_run_with_default_credentials(self):
-        self.create_task(credentials='{}').run()
+        list(self.create_task(credentials='{}').run())
 
     @with_luigi_config('database-export', 'database', 'foobar')
     def test_parameters_from_config(self):
@@ -113,7 +115,9 @@ def test_parameters_from_config(self):
         self.assertEquals(t.database, 'foobar')
 
     def test_run(self):
-        self.create_task().run()
+        task = self.create_task()
+        list(task.run())
+
         self.assertTrue(self.mock_mysql_connector.connect().cursor().execute.called)
         self.assertFalse(self.mock_mysql_connector.connect().rollback.called)
         self.assertTrue(self.mock_mysql_connector.connect().commit.called)
@@ -123,7 +127,8 @@ def test_run_with_failure(self):
         task = self.create_task()
         task.output().touch = MagicMock(side_effect=Exception("Failed to update marker"))
         with self.assertRaises(Exception):
-            task.run()
+            list(task.run())
+
         self.assertTrue(self.mock_mysql_connector.connect().cursor().execute.called)
         self.assertTrue(self.mock_mysql_connector.connect().rollback.called)
         self.assertFalse(self.mock_mysql_connector.connect().commit.called)
@@ -231,7 +236,7 @@ def test_insert_row_to_predefined_table(self):
     @with_luigi_config(('database-export', 'database', 'foobar'))
     def test_create_database(self):
         task = self.create_task()
-        task.run()
+        list(task.run())
 
         mock_cursor = self.mock_mysql_connector.connect.return_value.cursor.return_value
         mock_cursor.execute.assert_has_calls([
diff --git a/edx/analytics/tasks/common/tests/test_pathutil.py b/edx/analytics/tasks/common/tests/test_pathutil.py
index cc58febc35..60802b4669 100644
--- a/edx/analytics/tasks/common/tests/test_pathutil.py
+++ b/edx/analytics/tasks/common/tests/test_pathutil.py
@@ -59,9 +59,9 @@ class PathSelectionByDateIntervalTaskTest(unittest.TestCase):
     COMPLETE_SOURCE_PATHS = COMPLETE_SOURCE_PATHS_1 + COMPLETE_SOURCE_PATHS_2
     SOURCE = [SOURCE_1, SOURCE_2]
 
-    @patch('edx.analytics.tasks.util.s3_util.connect_s3')
+    @patch('luigi.contrib.s3.S3Client.s3')
     def test_requires(self, connect_s3_mock):
-        s3_conn_mock = connect_s3_mock.return_value
+        s3_conn_mock = connect_s3_mock
         bucket_mock = s3_conn_mock.get_bucket.return_value
 
         class FakeKey(object):
@@ -96,8 +96,8 @@ def test_default_source(self):
     def test_default_pattern(self):
         task = PathSelectionByDateIntervalTask(interval=Month.parse('2014-03'))
         self.assertEquals(task.pattern, (
-            r'.*tracking.log-(?P<date>\d{8}).*\.gz',
-            r'.*tracking.notalog-(?P<date>\d{8}).*\.gz',
+            r'.*tracking.log-(?P<date>\\d{8}).*\\.gz',
+            r'.*tracking.notalog-(?P<date>\\d{8}).*\\.gz',
         ))
 
     def test_filtering_of_urls(self):
@@ -192,14 +192,14 @@ def test_expanded_interval(self):
             'FakeServerGroup/tracking.log-20140401-1396379384.gz',
         ])
 
-    @with_luigi_config('event-logs', 'pattern', 'foobar')
+    @with_luigi_config('event-logs', 'pattern', '["foobar"]')
     def test_pattern_from_config(self):
         task = PathSelectionByDateIntervalTask(
             interval=Month.parse('2014-03')
         )
         self.assertEquals(task.pattern, ('foobar',))
 
-    @with_luigi_config('event-logs', 'pattern', ['foobar'])
+    @with_luigi_config('event-logs', 'pattern', '["foobar"]')
     def test_pattern_override(self):
         task = PathSelectionByDateIntervalTask(
             interval=Month.parse('2014-03'),
diff --git a/edx/analytics/tasks/common/tests/test_sqoop.py b/edx/analytics/tasks/common/tests/test_sqoop.py
index e8f3cce90d..43555b992c 100644
--- a/edx/analytics/tasks/common/tests/test_sqoop.py
+++ b/edx/analytics/tasks/common/tests/test_sqoop.py
@@ -21,7 +21,7 @@ def setUp(self):
         self.addCleanup(patcher.stop)
         self.mock_sqoop_password_target().path = "/temp/password_file"
 
-        patcher2 = patch("luigi.hadoop.run_and_track_hadoop_job")
+        patcher2 = patch("luigi.contrib.hadoop.run_and_track_hadoop_job")
         self.mock_run = patcher2.start()
         self.addCleanup(patcher2.stop)
 
diff --git a/edx/analytics/tasks/common/vertica_load.py b/edx/analytics/tasks/common/vertica_load.py
index 3ae9cc5336..50af2bc576 100644
--- a/edx/analytics/tasks/common/vertica_load.py
+++ b/edx/analytics/tasks/common/vertica_load.py
@@ -76,6 +76,12 @@ class VerticaCopyTask(VerticaCopyTaskMixin, luigi.Task):
     required_tasks = None
     output_target = None
 
+    restricted_roles = luigi.ListParameter(
+        config_path={'section': 'vertica-export', 'name': 'restricted_roles'},
+        default=[],
+        description='List of roles to which to provide access when a database column is marked as restricted.'
+    )
+
     def requires(self):
         if self.required_tasks is None:
             self.required_tasks = {
@@ -448,9 +454,8 @@ def restricted_columns(self):
     def create_access_policies(self, connection):
         cursor = connection.cursor()
         for column in self.restricted_columns:
-            restricted_roles_param = luigi.Parameter(is_list=True, config_path={'section': 'vertica-export', 'name': 'restricted_roles'}, default=[])
-            restricted_roles = ['dbadmin'] + list(restricted_roles_param.value)
-            expression = ' OR '.join(["ENABLED_ROLE('{0}')".format(role) for role in restricted_roles])
+            all_restricted_roles = ['dbadmin'] + list(self.restricted_roles)
+            expression = ' OR '.join(["ENABLED_ROLE('{0}')".format(role) for role in all_restricted_roles])
             statement = """
 CREATE ACCESS POLICY ON {schema}.{table} FOR COLUMN {column}
 CASE WHEN {expression} THEN {column}
@@ -602,8 +607,7 @@ class SchemaManagementTask(VerticaCopyTaskMixin, luigi.Task):
 
     date = luigi.DateParameter()
 
-    roles = luigi.Parameter(
-        is_list=True,
+    roles = luigi.ListParameter(
         config_path={'section': 'vertica-export', 'name': 'standard_roles'},
     )
 
diff --git a/edx/analytics/tasks/data_api/student_engagement.py b/edx/analytics/tasks/data_api/student_engagement.py
index 6b1dad0665..ee71f7cb3e 100644
--- a/edx/analytics/tasks/data_api/student_engagement.py
+++ b/edx/analytics/tasks/data_api/student_engagement.py
@@ -11,7 +11,7 @@
 from operator import itemgetter
 
 import luigi
-from luigi.hive import HiveQueryTask
+from luigi.contrib.hive import HiveQueryTask
 
 from edx.analytics.tasks.common.mapreduce import MapReduceJobTask, MapReduceJobTaskMixin, MultiOutputMapReduceJobTask
 from edx.analytics.tasks.common.mysql_load import MysqlInsertTask
diff --git a/edx/analytics/tasks/data_api/studentmodule_dist.py b/edx/analytics/tasks/data_api/studentmodule_dist.py
index e96a472efd..6acc84627c 100644
--- a/edx/analytics/tasks/data_api/studentmodule_dist.py
+++ b/edx/analytics/tasks/data_api/studentmodule_dist.py
@@ -43,7 +43,7 @@ class HistogramTaskFromSqoopParamsMixin(object):
         config_path={'section': 'database-export', 'name': 'credentials'},
         description='Credentials for the analytics db',
     )
-    sqoop_overwrite = luigi.BooleanParameter(  # prefixed with sqoop for disambiguation
+    sqoop_overwrite = luigi.BoolParameter(  # prefixed with sqoop for disambiguation
         default=False,
         description='Overwrite any existing imports.',
     )
diff --git a/edx/analytics/tasks/export/data_obfuscation.py b/edx/analytics/tasks/export/data_obfuscation.py
index 80d8ea6457..6d9261c44a 100644
--- a/edx/analytics/tasks/export/data_obfuscation.py
+++ b/edx/analytics/tasks/export/data_obfuscation.py
@@ -675,7 +675,7 @@ def requires(self):
 class DataObfuscationTask(ObfuscatorDownstreamMixin, luigi.WrapperTask):
     """Wrapper task for data obfuscation development."""
 
-    course = luigi.Parameter(is_list=True)
+    course = luigi.ListParameter()
     dump_root = luigi.Parameter()
     output_root = luigi.Parameter(
         config_path={'section': 'obfuscation', 'name': 'output_root'}
diff --git a/edx/analytics/tasks/export/database_exports.py b/edx/analytics/tasks/export/database_exports.py
index f59b286bfe..3846ae5911 100644
--- a/edx/analytics/tasks/export/database_exports.py
+++ b/edx/analytics/tasks/export/database_exports.py
@@ -130,7 +130,7 @@ class StudentModulePerCourseAfterImportWorkflow(StudentModulePerCourseTask):
         default=None,
         description='A "where" clause to be passed to Sqoop.',
     )
-    verbose = luigi.BooleanParameter(
+    verbose = luigi.BoolParameter(
         default=False,
         significant=False,
         description='Sqoop prints more information while working.',
diff --git a/edx/analytics/tasks/export/event_exports.py b/edx/analytics/tasks/export/event_exports.py
index 8651ea4bb6..e734200962 100644
--- a/edx/analytics/tasks/export/event_exports.py
+++ b/edx/analytics/tasks/export/event_exports.py
@@ -32,8 +32,7 @@ class EventExportTask(EventLogSelectionMixin, MultiOutputMapReduceJobTask):
         config_path={'section': 'event-export', 'name': 'config'},
         description='A URL to a YAML file that contains the list of organizations and servers to export events for.',
     )
-    org_id = luigi.Parameter(
-        is_list=True,
+    org_id = luigi.ListParameter(
         default=[],
         description='A list of organizations to process data for. If provided, only these organizations will be '
         'processed.  Otherwise, all valid organizations will be processed.',
diff --git a/edx/analytics/tasks/export/event_exports_by_course.py b/edx/analytics/tasks/export/event_exports_by_course.py
index 89b5bbd63e..90e2408f52 100644
--- a/edx/analytics/tasks/export/event_exports_by_course.py
+++ b/edx/analytics/tasks/export/event_exports_by_course.py
@@ -23,7 +23,7 @@ class EventExportByCourseTask(EventLogSelectionMixin, MultiOutputMapReduceJobTas
         config_path={'section': 'event-export-course', 'name': 'output_root'}
     )
 
-    course = luigi.Parameter(is_list=True, default=[])
+    course = luigi.ListParameter(default=[])
 
     def mapper(self, line):
         event, date_string = self.get_event_and_date_string(line) or (None, None)
diff --git a/edx/analytics/tasks/export/events_obfuscation.py b/edx/analytics/tasks/export/events_obfuscation.py
index 1b4588731f..1eb1000ff4 100644
--- a/edx/analytics/tasks/export/events_obfuscation.py
+++ b/edx/analytics/tasks/export/events_obfuscation.py
@@ -353,7 +353,7 @@ def extra_modules(self):
 class EventObfuscationTask(ObfuscatorDownstreamMixin, MapReduceJobTaskMixin, luigi.WrapperTask):
     """Wrapper task for course events obfuscation."""
 
-    course = luigi.Parameter(is_list=True)
+    course = luigi.ListParameter()
     dump_root = luigi.Parameter()
     output_root = luigi.Parameter()
     explicit_event_whitelist = luigi.Parameter(
diff --git a/edx/analytics/tasks/export/obfuscation.py b/edx/analytics/tasks/export/obfuscation.py
index e92566c038..0a189caffd 100644
--- a/edx/analytics/tasks/export/obfuscation.py
+++ b/edx/analytics/tasks/export/obfuscation.py
@@ -51,7 +51,7 @@ class ObfuscatedPackageTaskMixin(object):
     output_root = luigi.Parameter(
         config_path={'section': 'obfuscation', 'name': 'output_root'}
     )
-    recipient = luigi.Parameter(is_list=True)
+    recipient = luigi.ListParameter()
     format_version = luigi.Parameter()
 
 
@@ -165,7 +165,7 @@ def output(self):
 class MultiCourseObfuscatedCourseTask(ObfuscatedCourseTaskMixin, luigi.WrapperTask):
     """Task to obfuscate multiple courses at once."""
 
-    course = luigi.Parameter(is_list=True)
+    course = luigi.ListParameter()
 
     def requires(self):
         for course in self.course:   # pylint: disable=not-an-iterable
@@ -186,7 +186,7 @@ def requires(self):
 class MultiCourseObfuscatedPackageTask(ObfuscatedPackageTaskMixin, luigi.WrapperTask):
     """Task to package multiple courses at once."""
 
-    course = luigi.Parameter(is_list=True)
+    course = luigi.ListParameter()
     temporary_dir = luigi.Parameter(default=None)
 
     def requires(self):
diff --git a/edx/analytics/tasks/insights/answer_dist.py b/edx/analytics/tasks/insights/answer_dist.py
index f99a3ba848..0ee34e90de 100644
--- a/edx/analytics/tasks/insights/answer_dist.py
+++ b/edx/analytics/tasks/insights/answer_dist.py
@@ -10,7 +10,7 @@
 from operator import itemgetter
 
 import html5lib
-import luigi.s3
+import luigi
 from luigi.configuration import get_config
 
 import edx.analytics.tasks.util.eventlog as eventlog
@@ -599,15 +599,13 @@ class BaseAnswerDistributionDownstreamMixin(object):
         description='A unique identifier to distinguish one run from another.  It is used in '
         'the construction of output filenames, so each run will have distinct outputs.',
     )
-    src = luigi.Parameter(
-        is_list=True,
+    src = luigi.ListParameter(
         description='A list of URLs to the root location of input tracking log files.',
     )
     dest = luigi.Parameter(
         description='A URL to the root location to write output file(s).',
     )
-    include = luigi.Parameter(
-        is_list=True,
+    include = luigi.ListParameter(
         default=('*',),
         description='A list of patterns to be used to match input files, relative to `src` URL. '
         'The default value is [\'*\'].',
@@ -843,7 +841,7 @@ class AnswerDistributionToMySQLTaskWorkflow(
 ):
 
     # Override the parameter that normally defaults to false. This ensures that the table will always be overwritten.
-    overwrite = luigi.BooleanParameter(default=True, significant=False)
+    overwrite = luigi.BoolParameter(default=True, significant=False)
 
     @property
     def insert_source_task(self):
diff --git a/edx/analytics/tasks/insights/course_blocks.py b/edx/analytics/tasks/insights/course_blocks.py
index 37e1e6742c..2f5a7f85e5 100644
--- a/edx/analytics/tasks/insights/course_blocks.py
+++ b/edx/analytics/tasks/insights/course_blocks.py
@@ -169,7 +169,7 @@ class CourseBlocksApiDataTask(CourseBlocksDownstreamMixin, MapReduceJobTask):
         default='(Deleted block :)',
         description='Mark deleted (orphan) blocks with this string in course_path.',
     )
-    sort_orphan_blocks_up = luigi.BooleanParameter(
+    sort_orphan_blocks_up = luigi.BoolParameter(
         config_path={'section': 'course-blocks', 'name': 'sort_orphan_blocks_up'},
         default=False,
         description='If True, any deleted (orphan) blocks will be pushed to the top of the list '
diff --git a/edx/analytics/tasks/insights/course_list.py b/edx/analytics/tasks/insights/course_list.py
index eb70489d0f..268020af7d 100644
--- a/edx/analytics/tasks/insights/course_list.py
+++ b/edx/analytics/tasks/insights/course_list.py
@@ -54,7 +54,7 @@ class TimestampPartitionMixin(object):
 
     It can be used by HivePartitionTasks and tasks which invoke downstream HivePartitionTasks.
     """
-    date = luigi.DateParameter(
+    date = luigi.DateSecondParameter(
         default=datetime.datetime.utcnow(),
         description='Date/time for the data partition.  Default is UTC now.'
                     'Note that though this is a DateParameter, it also supports datetime objects, and so can '
diff --git a/edx/analytics/tasks/insights/database_imports.py b/edx/analytics/tasks/insights/database_imports.py
index 0bca90f6b6..d1c20d4998 100644
--- a/edx/analytics/tasks/insights/database_imports.py
+++ b/edx/analytics/tasks/insights/database_imports.py
@@ -6,7 +6,7 @@
 import textwrap
 
 import luigi
-from luigi.hive import HivePartitionTarget, HiveQueryTask
+from luigi.contrib.hive import HivePartitionTarget, HiveQueryTask
 
 from edx.analytics.tasks.common.sqoop import SqoopImportFromMysql, SqoopImportMixin
 from edx.analytics.tasks.util.hive import hive_database_name, hive_decimal_type
diff --git a/edx/analytics/tasks/insights/enrollments.py b/edx/analytics/tasks/insights/enrollments.py
index dda6138427..119ed2e590 100644
--- a/edx/analytics/tasks/insights/enrollments.py
+++ b/edx/analytics/tasks/insights/enrollments.py
@@ -205,13 +205,13 @@ def __init__(self, *args, **kwargs):
 
 class OverwriteHiveAndMysqlDownstreamMixin(object):
     """This mixin covers controls when we have both hive and mysql objects eligible for overwriting."""
-    overwrite_hive = luigi.BooleanParameter(
+    overwrite_hive = luigi.BoolParameter(
         default=False,
         description='Whether or not to overwrite the Hive intermediate objects; set to False by default.',
         significant=False
     )
 
-    overwrite_mysql = luigi.BooleanParameter(
+    overwrite_mysql = luigi.BoolParameter(
         default=False,
         description='Whether or not to overwrite the MySQL output objects; set to False by default.',
         significant=False
@@ -1474,7 +1474,7 @@ def indexes(self):
 class CourseSummaryEnrollmentDownstreamMixin(CourseEnrollmentDownstreamMixin, LoadInternalReportingCourseCatalogMixin):
     """Combines course enrollment and catalog parameters."""
 
-    enable_course_catalog = luigi.BooleanParameter(
+    enable_course_catalog = luigi.BoolParameter(
         config_path={'section': 'course-summary-enrollment', 'name': 'enable_course_catalog'},
         default=False,
         description="Enables course catalog data jobs."
diff --git a/edx/analytics/tasks/insights/location_per_course.py b/edx/analytics/tasks/insights/location_per_course.py
index 99ae243ad7..6bfa8e6297 100644
--- a/edx/analytics/tasks/insights/location_per_course.py
+++ b/edx/analytics/tasks/insights/location_per_course.py
@@ -7,7 +7,7 @@
 from collections import defaultdict
 
 import luigi
-from luigi.hive import HiveQueryTask
+from luigi.contrib.hive import HiveQueryTask
 
 from edx.analytics.tasks.common.mapreduce import MapReduceJobTask, MapReduceJobTaskMixin, MultiOutputMapReduceJobTask
 from edx.analytics.tasks.common.mysql_load import MysqlInsertTask
diff --git a/edx/analytics/tasks/insights/module_engagement.py b/edx/analytics/tasks/insights/module_engagement.py
index ad233f0418..8496aa172f 100644
--- a/edx/analytics/tasks/insights/module_engagement.py
+++ b/edx/analytics/tasks/insights/module_engagement.py
@@ -263,11 +263,11 @@ class ModuleEngagementMysqlTask(ModuleEngagementDownstreamMixin, IncrementalMysq
     Django ORM does not support composite primary key indexes, so we have to use a secondary index.
     """
 
-    allow_empty_insert = luigi.BooleanParameter(
+    allow_empty_insert = luigi.BoolParameter(
         default=False,
         config_path={'section': 'module-engagement', 'name': 'allow_empty_insert'},
     )
-    overwrite_hive = luigi.BooleanParameter(
+    overwrite_hive = luigi.BoolParameter(
         default=False,
         significant=False
     )
@@ -321,7 +321,7 @@ class ModuleEngagementIntervalTask(MapReduceJobTaskMixin, EventLogSelectionDowns
                                    OverwriteOutputMixin, OverwriteFromDateMixin, luigi.WrapperTask):
     """Compute engagement information over a range of dates and insert the results into Hive and MySQL"""
 
-    overwrite_mysql = luigi.BooleanParameter(
+    overwrite_mysql = luigi.BoolParameter(
         default=False,
         significant=False
     )
@@ -759,13 +759,13 @@ def data_task(self):
 class ModuleEngagementSummaryMetricRangesMysqlTask(ModuleEngagementDownstreamMixin, MysqlInsertTask):
     """Result store storage for the metric ranges."""
 
-    overwrite = luigi.BooleanParameter(
+    overwrite = luigi.BoolParameter(
         default=True,
         description='Overwrite the table when writing to it by default. Allow users to override this behavior if they '
                     'want.',
         significant=False
     )
-    allow_empty_insert = luigi.BooleanParameter(
+    allow_empty_insert = luigi.BoolParameter(
         default=False,
         config_path={'section': 'module-engagement', 'name': 'allow_empty_insert'},
     )
@@ -1228,7 +1228,7 @@ def requires(self):
 class ModuleEngagementRosterIndexDownstreamMixin(object):
     """Indexing parameters that can be specified at the workflow level."""
 
-    obfuscate = luigi.BooleanParameter(
+    obfuscate = luigi.BoolParameter(
         default=False,
         description='Generate fake names and email addresses for users. This can be used to generate production-like'
                     ' data sets that are more difficult to associate with particular users at a glance. Useful for'
@@ -1393,7 +1393,7 @@ class ModuleEngagementWorkflowTask(ModuleEngagementDownstreamMixin, ModuleEngage
     )
 
     # Don't use the OverwriteOutputMixin since it changes the behavior of complete() (which we don't want).
-    overwrite = luigi.BooleanParameter(default=False, significant=False)
+    overwrite = luigi.BoolParameter(default=False, significant=False)
     throttle = luigi.FloatParameter(
         config_path={'section': 'module-engagement', 'name': 'throttle'},
         description=ElasticsearchIndexTask.throttle.description,
diff --git a/edx/analytics/tasks/insights/problem_response.py b/edx/analytics/tasks/insights/problem_response.py
index da8b672147..fc80b11316 100644
--- a/edx/analytics/tasks/insights/problem_response.py
+++ b/edx/analytics/tasks/insights/problem_response.py
@@ -711,7 +711,7 @@ class ProblemResponseReportWorkflow(ProblemResponseTableMixin,
         description='URL directory where a marker file will be written on task completion.'
                     ' Note that the report task will not run if this marker file exists.',
     )
-    overwrite = luigi.BooleanParameter(
+    overwrite = luigi.BoolParameter(
         default=False,
         description='Set to True to force rebuild hive data and reports from tracking logs.'
     )
diff --git a/edx/analytics/tasks/insights/tags_dist.py b/edx/analytics/tasks/insights/tags_dist.py
index fdf5a7a739..a85d107849 100644
--- a/edx/analytics/tasks/insights/tags_dist.py
+++ b/edx/analytics/tasks/insights/tags_dist.py
@@ -3,7 +3,7 @@
 """
 import logging
 
-import luigi.s3
+import luigi
 
 import edx.analytics.tasks.util.eventlog as eventlog
 import edx.analytics.tasks.util.opaque_key_util as opaque_key_util
@@ -157,7 +157,7 @@ class TagsDistributionWorkflow(
     """
 
     # Override the parameter that normally defaults to false. This ensures that the table will always be overwritten.
-    overwrite = luigi.BooleanParameter(
+    overwrite = luigi.BoolParameter(
         default=True,
         description="Whether or not to overwrite existing outputs",
         significant=False
diff --git a/edx/analytics/tasks/insights/tests/test_answer_dist.py b/edx/analytics/tasks/insights/tests/test_answer_dist.py
index 991ffd67ff..b3691e233b 100644
--- a/edx/analytics/tasks/insights/tests/test_answer_dist.py
+++ b/edx/analytics/tasks/insights/tests/test_answer_dist.py
@@ -929,10 +929,10 @@ def setUp(self):
 
         self.task = AnswerDistributionOneFilePerCourseTask(
             mapreduce_engine='local',
-            src=None,
+            src=[],
             dest=None,
             name=None,
-            include=None,
+            include=[],
             output_root=None,
         )
 
@@ -968,10 +968,10 @@ def test_output_path_for_legacy_key(self):
         hashed_course_id = hashlib.sha1(course_id).hexdigest()
         task = AnswerDistributionOneFilePerCourseTask(
             mapreduce_engine='local',
-            src=None,
+            src=[],
             dest=None,
             name='name',
-            include=None,
+            include=[],
             output_root='/tmp',
         )
         output_path = task.output_path_for_key(course_id)
@@ -983,10 +983,10 @@ def test_output_path_for_opaque_key(self):
         hashed_course_id = hashlib.sha1(course_id).hexdigest()
         task = AnswerDistributionOneFilePerCourseTask(
             mapreduce_engine='local',
-            src=None,
+            src=[],
             dest=None,
             name='name',
-            include=None,
+            include=[],
             output_root='/tmp',
         )
         output_path = task.output_path_for_key(course_id)
@@ -1014,10 +1014,10 @@ def test_no_delete_output_root(self):
         self.assertTrue(os.path.exists(self.output_root))
         AnswerDistributionOneFilePerCourseTask(
             mapreduce_engine='local',
-            src=None,
+            src=[],
             dest=None,
             name='name',
-            include=None,
+            include=[],
             output_root=self.output_root,
         )
         self.assertTrue(os.path.exists(self.output_root))
@@ -1027,10 +1027,10 @@ def test_delete_output_root(self):
         # to get rid of the output_root directory.
         task = AnswerDistributionOneFilePerCourseTask(
             mapreduce_engine='local',
-            src=None,
+            src=[],
             dest=None,
             name='name',
-            include=None,
+            include=[],
             output_root=self.output_root,
             delete_output_root="true",
             marker=self.output_root,
diff --git a/edx/analytics/tasks/insights/tests/test_problem_response.py b/edx/analytics/tasks/insights/tests/test_problem_response.py
index fe5687853f..6504bae622 100644
--- a/edx/analytics/tasks/insights/tests/test_problem_response.py
+++ b/edx/analytics/tasks/insights/tests/test_problem_response.py
@@ -614,14 +614,18 @@ def assert_partition_value(self):
 
     def test_partition_value_with_start_end(self):
         self.create_task(
-            interval_start='2013-05-30',
+            interval_start=datetime.strptime('2013-05-30', '%Y-%m-%d'),
             interval_end=self.timestamp,
             partition_format=self.partition_format,
         )
         self.assert_partition_value()
 
-    def test_partition_value_with_interval(self):
-        interval = luigi.date_interval.Custom.parse('2013-05-30-{}'.format(self.timestamp.isoformat()))
+    def test_partition_value_with_no_interval(self):
+        # interval = luigi.date_interval.Custom.parse('2013-05-30-{}'.format(self.timestamp.isoformat()))
+        # TODO: fix this test to actually check what it's supposed to.
+        # The interval calculated using Custom.parse has always been None for strings with more than just a date range,
+        # so the task below has always used default values.
+        interval = None
         self.create_task(
             interval=interval,
             partition_format=self.partition_format,
diff --git a/edx/analytics/tasks/insights/user_activity.py b/edx/analytics/tasks/insights/user_activity.py
index b5ae5e6292..67f55eb8cf 100644
--- a/edx/analytics/tasks/insights/user_activity.py
+++ b/edx/analytics/tasks/insights/user_activity.py
@@ -310,14 +310,14 @@ class InsertToMysqlCourseActivityTask(WeeklyIntervalMixin, UserActivityDownstrea
     Creates/populates the `course_activity` Result store table.
     """
 
-    overwrite_hive = luigi.BooleanParameter(
+    overwrite_hive = luigi.BoolParameter(
         default=False,
         description='Overwrite the hive data used as source for this task. Users should set this to True '
                     'when using a persistent Hive metastore.',
         significant=False
     )
 
-    overwrite_mysql = luigi.BooleanParameter(
+    overwrite_mysql = luigi.BoolParameter(
         default=False,
         description='Overwrite the table if set to True. Allow users to override this behavior if they '
                     'want.',
diff --git a/edx/analytics/tasks/insights/video.py b/edx/analytics/tasks/insights/video.py
index ba6e475c3a..36c262496e 100644
--- a/edx/analytics/tasks/insights/video.py
+++ b/edx/analytics/tasks/insights/video.py
@@ -11,7 +11,7 @@
 import ciso8601
 import luigi
 from luigi import configuration
-from luigi.hive import HiveQueryTask
+from luigi.contrib.hive import HiveQueryTask
 from luigi.parameter import DateIntervalParameter
 
 from edx.analytics.tasks.common.mapreduce import MapReduceJobTask, MapReduceJobTaskMixin, MultiOutputMapReduceJobTask
@@ -883,13 +883,13 @@ def on_success(self):  # pragma: no cover
 class InsertToMysqlVideoTimelineTask(VideoTableDownstreamMixin, MysqlInsertTask):
     """Insert information about video timelines from a Hive table into MySQL."""
 
-    overwrite = luigi.BooleanParameter(
+    overwrite = luigi.BoolParameter(
         default=True,
         description='Overwrite the table when writing to it by default. Allow users to override this behavior if they '
                     'want.',
         significant=False
     )
-    allow_empty_insert = luigi.BooleanParameter(
+    allow_empty_insert = luigi.BoolParameter(
         default=False,
         description='Allow the video table to be empty (e.g. if no video activity has occurred)',
         config_path={'section': 'videos', 'name': 'allow_empty_insert'},
@@ -1042,13 +1042,13 @@ def on_success(self):  # pragma: no cover
 class InsertToMysqlVideoTask(VideoTableDownstreamMixin, MysqlInsertTask):
     """Insert summary information into the video table in MySQL."""
 
-    overwrite = luigi.BooleanParameter(
+    overwrite = luigi.BoolParameter(
         default=True,
         description='Overwrite the table when writing to it by default. Allow users to override this behavior if they '
                     'want.',
         significant=False
     )
-    allow_empty_insert = luigi.BooleanParameter(
+    allow_empty_insert = luigi.BoolParameter(
         default=False,
         description='Allow the video table to be empty (e.g. if no video activity has occurred)',
         config_path={'section': 'videos', 'name': 'allow_empty_insert'},
diff --git a/edx/analytics/tasks/launchers/local.py b/edx/analytics/tasks/launchers/local.py
index a510f58570..c78d513bfc 100644
--- a/edx/analytics/tasks/launchers/local.py
+++ b/edx/analytics/tasks/launchers/local.py
@@ -23,7 +23,7 @@
 import idna
 import luigi
 import luigi.configuration
-import luigi.hadoop
+import luigi.contrib.hadoop
 import opaque_keys
 import pyinstrument
 import requests
@@ -90,12 +90,12 @@ def main():
     #   - dependencies of opaque_keys:  bson, stevedore
     # - requests has several dependencies:
     #   - chardet, urllib3, certifi, idna
-    luigi.hadoop.attach(edx.analytics.tasks)
-    luigi.hadoop.attach(boto, cjson, filechunkio, opaque_keys, bson, stevedore, ciso8601, chardet, urllib3, certifi, idna, requests)
+    luigi.contrib.hadoop.attach(edx.analytics.tasks)
+    luigi.contrib.hadoop.attach(boto, cjson, filechunkio, opaque_keys, bson, stevedore, ciso8601, chardet, urllib3, certifi, idna, requests)
 
     if configuration.getboolean('ccx', 'enabled', default=False):
         import ccx_keys
-        luigi.hadoop.attach(ccx_keys)
+        luigi.contrib.hadoop.attach(ccx_keys)
 
     # TODO: setup logging for tasks or configured logging mechanism
 
diff --git a/edx/analytics/tasks/monitor/enrollment_validation.py b/edx/analytics/tasks/monitor/enrollment_validation.py
index 100ae3e96f..4a8ea9f7c7 100644
--- a/edx/analytics/tasks/monitor/enrollment_validation.py
+++ b/edx/analytics/tasks/monitor/enrollment_validation.py
@@ -45,7 +45,7 @@ class CourseEnrollmentValidationDownstreamMixin(EventLogSelectionDownstreamMixin
     )
 
     # Flag indicating whether to output synthetic events or tuples
-    tuple_output = luigi.BooleanParameter(
+    tuple_output = luigi.BoolParameter(
         default=False,
         description='A flag indicating that output should be in the form of tuples, not events. '
         'Default is False (output is events).',
@@ -53,7 +53,7 @@ class CourseEnrollmentValidationDownstreamMixin(EventLogSelectionDownstreamMixin
 
     # If set, generates events that occur before the start of the specified interval.
     # Default is incremental validation.
-    generate_before = luigi.BooleanParameter(
+    generate_before = luigi.BoolParameter(
         default=False,
         description='A flag indicating that events should be created preceding the '
         'specified interval. Default behavior is to suppress the generation of events '
@@ -62,7 +62,7 @@ class CourseEnrollmentValidationDownstreamMixin(EventLogSelectionDownstreamMixin
 
     # If set, events are included for transitions that don't result in a
     # change in enrollment state.  (For example, two activations in a row.)
-    include_nonstate_changes = luigi.BooleanParameter(
+    include_nonstate_changes = luigi.BoolParameter(
         default=False,
         description='A flag indicating that events should be created '
         'to fix all transitions, even those that don\'t result in a change in enrollment '
diff --git a/edx/analytics/tasks/tests/acceptance/__init__.py b/edx/analytics/tasks/tests/acceptance/__init__.py
index 66059e3015..23677facd1 100644
--- a/edx/analytics/tasks/tests/acceptance/__init__.py
+++ b/edx/analytics/tasks/tests/acceptance/__init__.py
@@ -112,6 +112,14 @@ def modify_target_for_local_server(target):
         return target
 
 
+def as_list_param(value, escape_quotes=True):
+    """Convenience method to convert a single string to a format expected by a Luigi ListParameter."""
+    if escape_quotes:
+        return '[\\"{}\\"]'.format(value)
+    else:
+        return json.dumps([value, ])
+
+
 def coerce_columns_to_string(row):
     # Vertica response includes datatypes in some columns i-e. datetime, Decimal etc. so convert
     # them into string before comparison with expected output.
@@ -201,7 +209,7 @@ def setUp(self):
             },
             'manifest': {
                 'path': url_path_join(self.test_root, 'manifest'),
-                'lib_jar': self.config['oddjob_jar']
+                'lib_jar': self.config['oddjob_jar'],
             },
             'database-import': {
                 'credentials': self.config['credentials_file_url'],
@@ -223,10 +231,12 @@ def setUp(self):
                 'geolocation_data': self.config['geolocation_data']
             },
             'event-logs': {
-                'source': self.test_src
+                'source': as_list_param(self.test_src, escape_quotes=False),
+                'pattern': as_list_param(".*tracking.log-(?P<date>\\d{8}).*\\.gz", escape_quotes=False),
             },
             'segment-logs': {
-                'source': self.test_src
+                'source': as_list_param(self.test_src, escape_quotes=False),
+                'pattern': as_list_param(".*segment.log-(?P<date>\\d{8}).*\\.gz", escape_quotes=False),
             },
             'course-structure': {
                 'api_root_url': 'acceptance.test',
@@ -264,7 +274,7 @@ def setUp(self):
                 'schema': schema
             }
         if 'elasticsearch_host' in self.config:
-            task_config_override['elasticsearch']['host'] = self.config['elasticsearch_host']
+            task_config_override['elasticsearch']['host'] = as_list_param(self.config['elasticsearch_host'], escape_quotes=False)
         if 'elasticsearch_connection_class' in self.config:
             task_config_override['elasticsearch']['connection_type'] = self.config['elasticsearch_connection_class']
         if 'manifest_input_format' in self.config:
diff --git a/edx/analytics/tasks/tests/acceptance/test_answer_dist.py b/edx/analytics/tasks/tests/acceptance/test_answer_dist.py
index 9cba64220e..286d2f7687 100644
--- a/edx/analytics/tasks/tests/acceptance/test_answer_dist.py
+++ b/edx/analytics/tasks/tests/acceptance/test_answer_dist.py
@@ -5,7 +5,7 @@
 import logging
 import os
 
-from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase
+from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase, as_list_param
 from edx.analytics.tasks.util.url import url_path_join
 
 log = logging.getLogger(__name__)
@@ -42,14 +42,14 @@ class AnswerDistributionAcceptanceTest(BaseAnswerDistributionAcceptanceTest):
     def test_answer_distribution(self):
         self.task.launch([
             'AnswerDistributionOneFilePerCourseTask',
-            '--src', self.test_src,
+            '--src', as_list_param(self.test_src),
             '--dest', url_path_join(self.test_root, 'dst'),
             '--name', 'test',
             '--output-root', self.test_out,
-            '--include', '"*"',
+            '--include', as_list_param('"*"'),
             '--manifest', url_path_join(self.test_root, 'manifest.txt'),
             '--base-input-format', self.input_format,
-            '--lib-jar', self.oddjob_jar,
+            '--lib-jar', as_list_param(self.oddjob_jar),
             '--n-reduce-tasks', str(self.NUM_REDUCERS),
         ])
         self.validate_output()
@@ -79,13 +79,13 @@ class AnswerDistributionMysqlAcceptanceTests(BaseAnswerDistributionAcceptanceTes
     def test_answer_distribution_mysql(self):
         self.task.launch([
             'AnswerDistributionToMySQLTaskWorkflow',
-            '--src', self.test_src,
+            '--src', as_list_param(self.test_src),
             '--dest', url_path_join(self.test_root, 'dst'),
             '--name', 'test',
-            '--include', '"*"',
+            '--include', as_list_param('"*"'),
             '--manifest', url_path_join(self.test_root, 'manifest.txt'),
             '--base-input-format', self.input_format,
-            '--lib-jar', self.oddjob_jar,
+            '--lib-jar', as_list_param(self.oddjob_jar),
             '--n-reduce-tasks', str(self.NUM_REDUCERS),
             '--credentials', self.export_db.credentials_file_url,
         ])
diff --git a/edx/analytics/tasks/tests/acceptance/test_course_blocks.py b/edx/analytics/tasks/tests/acceptance/test_course_blocks.py
index ca4d1b278c..7dc14af200 100644
--- a/edx/analytics/tasks/tests/acceptance/test_course_blocks.py
+++ b/edx/analytics/tasks/tests/acceptance/test_course_blocks.py
@@ -56,7 +56,7 @@ def test_partition_task_hive_input(self):
 
     def validate_partition_task(self):
         """Run the CourseBlocksPartitionTask and test its output."""
-        date = self.DATE.strftime('%Y-%m-%d')
+        date = self.DATE.strftime('%Y-%m-%dT%H%M%S')
         input_root = url_path_join(self.warehouse_path, 'course_list', self.partition)
 
         self.task.launch([
diff --git a/edx/analytics/tasks/tests/acceptance/test_course_list.py b/edx/analytics/tasks/tests/acceptance/test_course_list.py
index 24f19a322c..588672f589 100644
--- a/edx/analytics/tasks/tests/acceptance/test_course_list.py
+++ b/edx/analytics/tasks/tests/acceptance/test_course_list.py
@@ -31,7 +31,7 @@ def setUp(self):
 
     def test_partition_task(self):
         """Run the CourseListPartitionTask and test its output."""
-        date = self.DATE.strftime('%Y-%m-%d')
+        date = self.DATE.strftime('%Y-%m-%dT%H%M%S')
         self.task.launch([
             'CourseListPartitionTask',
             '--date', date,
diff --git a/edx/analytics/tasks/tests/acceptance/test_enrollment_validation.py b/edx/analytics/tasks/tests/acceptance/test_enrollment_validation.py
index 8402bbcb30..4cdcfceb16 100644
--- a/edx/analytics/tasks/tests/acceptance/test_enrollment_validation.py
+++ b/edx/analytics/tasks/tests/acceptance/test_enrollment_validation.py
@@ -7,7 +7,7 @@
 import StringIO
 from collections import defaultdict
 
-from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase
+from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase, as_list_param
 from edx.analytics.tasks.util.url import url_path_join
 
 log = logging.getLogger(__name__)
@@ -64,8 +64,8 @@ def launch_task(self, output_root, extra_source=None, run_with_validation_events
 
         # Widen the interval to include the latest validation events.
         interval = self.WIDER_DATE_INTERVAL if run_with_validation_events else self.DATE_INTERVAL
-        source_pattern = r'".*?\.log-(?P<date>\d{8}).*\.gz"'
-        validation_pattern = r'".*?enroll_validated_(?P<date>\d{8})\.log\.gz"'
+        source_pattern = '[\\".*?.log-.*.gz\\"]'
+        validation_pattern = '".*?enroll_validated_\d{8}\.log\.gz"'
         launch_args = [
             'EnrollmentValidationWorkflow',
             '--interval', interval,
@@ -73,15 +73,15 @@ def launch_task(self, output_root, extra_source=None, run_with_validation_events
             '--validation-pattern', validation_pattern,
             '--credentials', self.import_db.credentials_file_url,
             '--n-reduce-tasks', str(self.NUM_REDUCERS),
-            '--source', self.test_src,
             '--pattern', source_pattern,
             '--output-root', output_root,
         ]
         # An extra source means we're using synthetic events, so we
         # don't want to generate outside the interval in that case.
         if extra_source:
-            launch_args.extend(['--source', extra_source])
+            launch_args.extend(['--source', '[\\"{}\\",\\"{}\\"]'.format(self.test_src, extra_source)])
         else:
+            launch_args.extend(['--source', as_list_param(self.test_src)])
             launch_args.extend(['--generate-before'])
         if run_with_validation_events:
             launch_args.extend(['--expected-validation', "{}T00".format(self.END_DATE)])
diff --git a/edx/analytics/tasks/tests/acceptance/test_event_export.py b/edx/analytics/tasks/tests/acceptance/test_event_export.py
index f8c8972b76..6a410f59b2 100644
--- a/edx/analytics/tasks/tests/acceptance/test_event_export.py
+++ b/edx/analytics/tasks/tests/acceptance/test_event_export.py
@@ -8,7 +8,7 @@
 import tempfile
 import textwrap
 
-from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase
+from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase, as_list_param
 from edx.analytics.tasks.tests.acceptance.services import fs
 from edx.analytics.tasks.util.url import url_path_join
 
@@ -84,7 +84,7 @@ def test_event_log_exports_using_manifest(self):
         for environment in ['prod', 'edge']:
             self.task.launch([
                 'EventExportTask',
-                '--source', url_path_join(self.test_src, environment),
+                '--source', as_list_param(url_path_join(self.test_src, environment)),
                 '--output-root', self.test_out,
                 '--config', self.test_config,
                 '--environment', environment,
diff --git a/edx/analytics/tasks/tests/acceptance/test_financial_reports.py b/edx/analytics/tasks/tests/acceptance/test_financial_reports.py
index efa1211b98..8ce9f30480 100644
--- a/edx/analytics/tasks/tests/acceptance/test_financial_reports.py
+++ b/edx/analytics/tasks/tests/acceptance/test_financial_reports.py
@@ -55,7 +55,9 @@ def test_end_to_end(self):
             '--n-reduce-tasks', str(self.NUM_REDUCERS),
         ])
 
-        final_output_task = LoadInternalReportingOrderTransactionsToWarehouse(import_date=self.UPPER_BOUND_DATE)
+        final_output_task = LoadInternalReportingOrderTransactionsToWarehouse(
+            import_date=luigi.DateParameter().parse(self.UPPER_BOUND_DATE)
+        )
         columns = [x[0] for x in final_output_task.columns]
 
         with self.vertica.cursor() as cursor:
diff --git a/edx/analytics/tasks/tests/acceptance/test_location_per_course.py b/edx/analytics/tasks/tests/acceptance/test_location_per_course.py
index 829b433ff5..c2d551fa3f 100644
--- a/edx/analytics/tasks/tests/acceptance/test_location_per_course.py
+++ b/edx/analytics/tasks/tests/acceptance/test_location_per_course.py
@@ -3,7 +3,7 @@
 
 from luigi.date_interval import Date
 
-from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase, when_geolocation_data_available
+from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase, as_list_param, when_geolocation_data_available
 
 
 class LocationByCourseAcceptanceTest(AcceptanceTestCase):
@@ -28,7 +28,7 @@ def test_location_by_course(self):
 
         self.task.launch([
             'InsertToMysqlCourseEnrollByCountryWorkflow',
-            '--source', self.test_src,
+            '--source', as_list_param(self.test_src),
             '--interval', self.DATE_INTERVAL.to_string(),
             '--n-reduce-tasks', str(self.NUM_REDUCERS),
         ])
diff --git a/edx/analytics/tasks/tests/acceptance/test_obfuscation.py b/edx/analytics/tasks/tests/acceptance/test_obfuscation.py
index e2190498cf..0d79bcd9de 100644
--- a/edx/analytics/tasks/tests/acceptance/test_obfuscation.py
+++ b/edx/analytics/tasks/tests/acceptance/test_obfuscation.py
@@ -8,7 +8,7 @@
 import tarfile
 import tempfile
 
-from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase, when_geolocation_data_available
+from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase, as_list_param, when_geolocation_data_available
 from edx.analytics.tasks.tests.acceptance.services import fs, shell
 from edx.analytics.tasks.util.file_util import copy_file_to_file
 from edx.analytics.tasks.util.opaque_key_util import get_filename_safe_course_id
@@ -104,7 +104,7 @@ def run_obfuscated_package_task(self):
             '--gpg-key-dir', self.test_gpg_key_dir,
             '--gpg-master-key', 'daemon+master@edx.org',
             '--output-root', self.test_out,
-            '--recipient', 'daemon@edx.org',
+            '--recipient', as_list_param('daemon@edx.org'),
             '--format-version', self.FORMAT_VERSION
         ])
 
diff --git a/edx/analytics/tasks/tests/acceptance/test_problem_response.py b/edx/analytics/tasks/tests/acceptance/test_problem_response.py
index ba712621c9..6ad86eae8d 100644
--- a/edx/analytics/tasks/tests/acceptance/test_problem_response.py
+++ b/edx/analytics/tasks/tests/acceptance/test_problem_response.py
@@ -66,7 +66,7 @@ def test_problem_response_report_hive_input(self):
     def validate_problem_response_report(self):
         """Run the ProblemResponseReportWorkflow task and test the output."""
         marker_path = url_path_join(self.test_out, 'marker-{}'.format(str(time.time())))
-        report_date = self.DATE.strftime('%Y-%m-%d')
+        report_date = self.DATE.strftime('%Y-%m-%dT%H%M%S')
 
         # The test tracking.log file contains problem_check events for 2016-09-06, 09-07, and 09-08.
         # However, to test the interval parameter propagation, we deliberately exclude all but the 2016-09-07 events.
diff --git a/edx/analytics/tasks/tests/acceptance/test_student_engagement.py b/edx/analytics/tasks/tests/acceptance/test_student_engagement.py
index 0441fb512f..eab1aa9d79 100644
--- a/edx/analytics/tasks/tests/acceptance/test_student_engagement.py
+++ b/edx/analytics/tasks/tests/acceptance/test_student_engagement.py
@@ -11,7 +11,7 @@
 from pandas import read_csv
 from pandas.util.testing import assert_frame_equal
 
-from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase
+from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase, as_list_param
 from edx.analytics.tasks.util.url import url_path_join
 
 log = logging.getLogger(__name__)
@@ -99,7 +99,7 @@ def run_task(self, interval_type):
         """Run the CSV-generating task."""
         self.task.launch([
             'StudentEngagementCsvFileTask',
-            '--source', self.test_src,
+            '--source', as_list_param(self.test_src),
             '--output-root', url_path_join(self.test_out, interval_type),
             '--n-reduce-tasks', str(self.NUM_REDUCERS),
             '--interval', self.interval,
@@ -193,7 +193,7 @@ def test_forum_engagement(self):
     def run_and_check(self, interval_type):
         self.task.launch([
             'StudentEngagementToMysqlTask',
-            '--source', self.test_src,
+            '--source', as_list_param(self.test_src),
             '--credentials', self.export_db.credentials_file_url,
             '--n-reduce-tasks', str(self.NUM_REDUCERS),
             '--interval', '2015-09-01-2015-09-16',
diff --git a/edx/analytics/tasks/tests/acceptance/test_tags_dist.py b/edx/analytics/tasks/tests/acceptance/test_tags_dist.py
index 9db67ce77c..cacb367ced 100644
--- a/edx/analytics/tasks/tests/acceptance/test_tags_dist.py
+++ b/edx/analytics/tasks/tests/acceptance/test_tags_dist.py
@@ -4,7 +4,7 @@
 
 import datetime
 
-from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase
+from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase, as_list_param
 from edx.analytics.tasks.util.url import url_path_join
 
 
@@ -19,7 +19,7 @@ def test_base(self):
 
         self.task.launch([
             'TagsDistributionWorkflow',
-            '--source', self.test_src,
+            '--source', as_list_param(self.test_src),
             '--interval', '2010-01-01-2020-01-01',
             '--n-reduce-tasks', str(self.NUM_REDUCERS),
             '--output-root', url_path_join(self.test_out, 'tags_dist_acceptance', ''),
diff --git a/edx/analytics/tasks/tests/acceptance/test_user_activity.py b/edx/analytics/tasks/tests/acceptance/test_user_activity.py
index 7258e3da16..6f9cc909fd 100644
--- a/edx/analytics/tasks/tests/acceptance/test_user_activity.py
+++ b/edx/analytics/tasks/tests/acceptance/test_user_activity.py
@@ -2,7 +2,7 @@
 
 import datetime
 
-from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase
+from edx.analytics.tasks.tests.acceptance import AcceptanceTestCase, as_list_param
 
 
 class UserActivityAcceptanceTest(AcceptanceTestCase):
@@ -21,7 +21,7 @@ def test_user_activity(self):
 
         self.task.launch([
             'InsertToMysqlCourseActivityTask',
-            '--source', self.test_src,
+            '--source', as_list_param(self.test_src),
             '--end-date', self.END_DATE.isoformat(),
             '--weeks', str(self.NUM_WEEKS),
             '--credentials', self.export_db.credentials_file_url,
diff --git a/edx/analytics/tasks/util/elasticsearch_target.py b/edx/analytics/tasks/util/elasticsearch_target.py
index 6064b0984f..39e61cc75d 100644
--- a/edx/analytics/tasks/util/elasticsearch_target.py
+++ b/edx/analytics/tasks/util/elasticsearch_target.py
@@ -6,7 +6,7 @@
 
 import luigi
 import luigi.configuration
-import luigi.hdfs
+from luigi.contrib.hdfs.target import HdfsTarget
 
 try:
     import elasticsearch
@@ -17,7 +17,7 @@
 log = logging.getLogger(__name__)
 
 
-class ElasticsearchTarget(luigi.hdfs.HdfsTarget):
+class ElasticsearchTarget(HdfsTarget):
     """
     Represents an index in an elasticsearch cluster.
 
diff --git a/edx/analytics/tasks/util/hive.py b/edx/analytics/tasks/util/hive.py
index 11adcab4b8..13c852893b 100644
--- a/edx/analytics/tasks/util/hive.py
+++ b/edx/analytics/tasks/util/hive.py
@@ -5,8 +5,8 @@
 
 import luigi
 from luigi.configuration import get_config
-from luigi.hive import HivePartitionTarget, HiveQueryRunner, HiveQueryTask, HiveTableTarget
-from luigi.parameter import BooleanParameter, Parameter
+from luigi.contrib.hive import HivePartitionTarget, HiveQueryRunner, HiveQueryTask, HiveTableTarget
+from luigi.parameter import BoolParameter, Parameter
 
 from edx.analytics.tasks.common.mysql_load import MysqlInsertTask
 from edx.analytics.tasks.util.overwrite import OverwriteOutputMixin
@@ -328,8 +328,12 @@ def requires(self):
         yield self.hive_table_task
 
     def output(self):
+        # Ugh.  A change in Luigi 1.0.22 (after our 1.0.17 fork) resulted in a change in ApacheHiveCommandClient.table_exists()
+        # behavior, so that it throws an exception when checking for a specific partition when the table doesn't exist.
+        # This means that HivePartitionTarget.exists() will fail, where before it succeeded even if the table did not exist.
+        # So change fail_missing_table=False here.  There is no reason for it anyway.
         return HivePartitionTarget(
-            self.hive_table_task.table, self.partition.as_dict(), database=hive_database_name(), fail_missing_table=True
+            self.hive_table_task.table, self.partition.as_dict(), database=hive_database_name(), fail_missing_table=False
         )
 
     def job_runner(self):
@@ -426,7 +430,7 @@ class OverwriteAwareHiveQueryDataTask(WarehouseMixin, OverwriteOutputMixin, Hive
     A generalized Data task whose output is a hive table populated from a hive query.
     """
 
-    overwrite_target_partition = BooleanParameter(
+    overwrite_target_partition = BoolParameter(
         significant=False,
         description='Overwrite the target partition, deleting any existing data.  This will not impact other '
                     'partitions.  Do not use with incrementally built partitions.',
diff --git a/edx/analytics/tasks/util/obfuscate_util.py b/edx/analytics/tasks/util/obfuscate_util.py
index 0073eb91bd..0d6e7f1942 100644
--- a/edx/analytics/tasks/util/obfuscate_util.py
+++ b/edx/analytics/tasks/util/obfuscate_util.py
@@ -153,7 +153,7 @@ def _initialize_user_info(self):
 class ObfuscatorDownstreamMixin(UserInfoDownstreamMixin):
     """Class for defining Luigi functions used downstream of obfuscating classes."""
 
-    entities = luigi.Parameter(is_list=True, default=[])
+    entities = luigi.ListParameter(default=[])
     log_context = luigi.IntParameter(default=None)
 
 
diff --git a/edx/analytics/tasks/util/overwrite.py b/edx/analytics/tasks/util/overwrite.py
index cc05f564ac..68a264a9a1 100644
--- a/edx/analytics/tasks/util/overwrite.py
+++ b/edx/analytics/tasks/util/overwrite.py
@@ -24,7 +24,7 @@ class OverwriteOutputMixin(object):
     Note that this should be included in a task definition *before*
     the Task base class, so that the complete() method is overridden.
     """
-    overwrite = luigi.BooleanParameter(
+    overwrite = luigi.BoolParameter(
         default=False,
         description='Whether or not to overwrite existing outputs; set to False by default for now.',
         significant=False
diff --git a/edx/analytics/tasks/util/s3_util.py b/edx/analytics/tasks/util/s3_util.py
index d73ae0404d..bc05d64fa5 100644
--- a/edx/analytics/tasks/util/s3_util.py
+++ b/edx/analytics/tasks/util/s3_util.py
@@ -11,8 +11,9 @@
 from boto import connect_s3
 from boto.s3.key import Key
 from filechunkio import FileChunkIO
-from luigi.hdfs import HdfsTarget, Plain
-from luigi.s3 import AtomicS3File, S3Client
+from luigi.contrib.hdfs.format import Plain
+from luigi.contrib.hdfs.target import HdfsTarget
+from luigi.contrib.s3 import AtomicS3File, S3Client
 
 log = logging.getLogger(__name__)
 
@@ -127,99 +128,20 @@ def func(name):
 
 class ScalableS3Client(S3Client):
     """
-    S3 client that adds support for multipart uploads and requires minimal permissions.
-
-    Uses S3 multipart upload API for large files, and regular S3 puts for smaller files.
-
-    This client should only require PutObject and PutObjectAcl permissions in order to write to the target bucket.
+    S3 client that adds support for defaulting host name.
     """
     # TODO: Make this behavior configurable and submit this change upstream.
 
     def __init__(self, aws_access_key_id=None, aws_secret_access_key=None, **kwargs):
-        # Note: this deliberately does not call the base class __init__ method,
-        # to avoid the connect_s3 call there without the host argument, which fails.
+
         if not aws_access_key_id:
             aws_access_key_id = self._get_s3_config('aws_access_key_id')
         if not aws_secret_access_key:
             aws_secret_access_key = self._get_s3_config('aws_secret_access_key')
         if 'host' not in kwargs:
             kwargs['host'] = self._get_s3_config('host') or 's3.amazonaws.com'
-        self.s3 = connect_s3(aws_access_key_id, aws_secret_access_key, is_secure=True, **kwargs)
-
-    def put(self, local_path, destination_s3_path):
-        """Put an object stored locally to an S3 path."""
-
-        # parse path into bucket and key
-        (bucket, key) = self._path_to_bucket_and_key(destination_s3_path)
-
-        # If Boto is passed "validate=True", it will require an
-        # additional permission to be present when asked to list all
-        # of the keys in the bucket.  We want to minimize the set of
-        # required permissions so we get a reference to the bucket
-        # without validating that it exists.  It should only require
-        # PutObject and PutObjectAcl permissions in order to write to
-        # the target bucket.
-        s3_bucket = self.s3.get_bucket(bucket, validate=False)
-
-        # Check first if we should be doing a multipart upload.
-        source_size_bytes = os.stat(local_path).st_size
-        if source_size_bytes < MULTIPART_UPLOAD_THRESHOLD:
-            self._upload_single(local_path, s3_bucket, key)
-        else:
-            log.info("File '%s' has size %d, exceeding threshold %d for using put -- using multipart upload.",
-                     destination_s3_path, source_size_bytes, MULTIPART_UPLOAD_THRESHOLD)
-            self._upload_multipart(local_path, destination_s3_path, s3_bucket, key, source_size_bytes)
-
-    def _upload_single(self, local_path, s3_bucket, key):
-        """
-        Write a local file to an S3 key using single PUT.
-
-        This only works for files < 5GB in size.
-        """
-        s3_key = Key(s3_bucket)
-        s3_key.key = key
-        # Explicitly set the ACL policy when putting the object, so
-        # that it has an ACL when AWS writes to keys from another account.
-        s3_key.set_contents_from_filename(local_path, policy=DEFAULT_KEY_ACCESS_POLICY)
-
-    def _upload_multipart(self, local_path, destination_s3_path, s3_bucket, key, source_size_bytes):
-        """Upload a large local file to an S3 path, using S3's multipart upload API."""
-
-        # Explicitly set the ACL policy when putting the object, so
-        # that it has an ACL when AWS writes to keys from another account.
-        multipart = s3_bucket.initiate_multipart_upload(key, policy=DEFAULT_KEY_ACCESS_POLICY)
-
-        number_of_chunks, bytes_per_chunk = self._get_chunk_specs(source_size_bytes)
-        log.info("Uploading file '%s' with size %d in %d parts, with chunksize of %d.",
-                 destination_s3_path, source_size_bytes, number_of_chunks, bytes_per_chunk)
-
-        chunk_generator = self._generate_chunks(source_size_bytes, number_of_chunks, bytes_per_chunk)
-        for part_num, chunk_byte_offset, num_bytes in chunk_generator:
-            with FileChunkIO(local_path, 'r', offset=chunk_byte_offset, bytes=num_bytes) as chunk:
-                multipart.upload_part_from_file(fp=chunk, part_num=part_num)
-
-        if len(multipart.get_all_parts()) == number_of_chunks:
-            multipart.complete_upload()
-        else:
-            multipart.cancel_upload()
-
-    def _get_chunk_specs(self, source_size_bytes):
-        """Returns number of chunks and bytes-per-chunk given a filesize."""
-        # Select a chunk size, so that the chunk size grows with the overall size, but
-        # more slowly.  (Scale so that it equals the minimum chunk size.)
-        bytes_per_chunk = int(math.sqrt(MINIMUM_BYTES_PER_CHUNK) * math.sqrt(source_size_bytes))
-        bytes_per_chunk = min(max(bytes_per_chunk, MINIMUM_BYTES_PER_CHUNK), MULTIPART_UPLOAD_THRESHOLD)
-        number_of_chunks = int(math.ceil(source_size_bytes / float(bytes_per_chunk)))
-        return number_of_chunks, bytes_per_chunk
-
-    def _generate_chunks(self, source_size_bytes, number_of_chunks, bytes_per_chunk):
-        """Returns the index, offset, and size of chunks."""
-        for chunk_index in range(number_of_chunks):
-            chunk_byte_offset = chunk_index * bytes_per_chunk
-            remaining_bytes_in_file = source_size_bytes - chunk_byte_offset
-            num_bytes = min([bytes_per_chunk, remaining_bytes_in_file])
-            # indexing of parts is one-based.
-            yield chunk_index + 1, chunk_byte_offset, num_bytes
+
+        super(ScalableS3Client, self).__init__(aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key, **kwargs)
 
 
 class S3HdfsTarget(HdfsTarget):
@@ -228,8 +150,6 @@ class S3HdfsTarget(HdfsTarget):
     # Luigi does not support writing to HDFS targets that point to complete URLs like "s3://foo/bar" it only supports
     # HDFS paths that look like standard file paths "/foo/bar".
 
-    # (This class also provides a customized implementation for S3Client.)
-
     # TODO: Fix the upstream bug in luigi that prevents writing to HDFS files that are specified by complete URLs
 
     def __init__(self, path=None, format=Plain, is_tmp=False):
@@ -245,4 +165,4 @@ def open(self, mode='r'):
             safe_path = self.path.replace('s3n://', 's3://')
             if not hasattr(self, 's3_client'):
                 self.s3_client = ScalableS3Client()
-            return AtomicS3File(safe_path, self.s3_client)
+            return AtomicS3File(safe_path, self.s3_client, policy=DEFAULT_KEY_ACCESS_POLICY)
diff --git a/edx/analytics/tasks/util/tests/test_manifest.py b/edx/analytics/tasks/util/tests/test_manifest.py
index 80699b8dcb..062c4063e3 100644
--- a/edx/analytics/tasks/util/tests/test_manifest.py
+++ b/edx/analytics/tasks/util/tests/test_manifest.py
@@ -3,7 +3,7 @@
 from unittest import TestCase
 
 import luigi
-import luigi.hdfs
+from luigi.contrib.hdfs.target import HdfsTarget
 from mock import patch
 
 from edx.analytics.tasks.util.manifest import (
@@ -49,7 +49,7 @@ def test_parameters_not_configured(self):
         self.assertFalse(hasattr(target, 'input_format'))
 
     def test_manifest_file_construction(self):
-        target = create_manifest_target(self.MANIFEST_ID, [luigi.hdfs.HdfsTarget('s3://foo/bar')])
+        target = create_manifest_target(self.MANIFEST_ID, [HdfsTarget('s3://foo/bar')])
         self.assertEquals(target.value, 's3://foo/bar\n')
 
     @with_luigi_config('manifest', 'threshold', 1)
diff --git a/edx/analytics/tasks/util/tests/test_s3_util.py b/edx/analytics/tasks/util/tests/test_s3_util.py
index b635541f08..dad8cde8e9 100644
--- a/edx/analytics/tasks/util/tests/test_s3_util.py
+++ b/edx/analytics/tasks/util/tests/test_s3_util.py
@@ -97,44 +97,3 @@ def test_generate_with_trailing_slash(self):
             (bucket_name, root.rstrip('/'), "subdir1/path1"),
             (bucket_name, root.rstrip('/'), "path2")
         ]))
-
-
-class ScalableS3ClientTestCase(TestCase):
-    """Tests for ScalableS3Client class."""
-
-    def setUp(self):
-        patcher = patch('edx.analytics.tasks.util.s3_util.connect_s3')
-        patcher.start()
-        self.addCleanup(patcher.stop)
-
-        self.client = s3_util.ScalableS3Client()
-
-    def _assert_get_chunk_specs(self, source_size_bytes, expected_num_chunks, expected_chunk_size):
-        """Asserts that _get_chunk_specs returns the expected values."""
-        number_of_chunks, bytes_per_chunk = self.client._get_chunk_specs(source_size_bytes)
-        self.assertEquals(number_of_chunks, expected_num_chunks)
-        self.assertEquals(bytes_per_chunk, expected_chunk_size)
-
-    def test_get_minimum_chunk_specs(self):
-        self._assert_get_chunk_specs(1, 1, s3_util.MINIMUM_BYTES_PER_CHUNK)
-        self._assert_get_chunk_specs(s3_util.MINIMUM_BYTES_PER_CHUNK, 1, s3_util.MINIMUM_BYTES_PER_CHUNK)
-
-    def test_get_maximum_chunk_specs(self):
-        size = ((s3_util.MULTIPART_UPLOAD_THRESHOLD * s3_util.MULTIPART_UPLOAD_THRESHOLD) /
-                s3_util.MINIMUM_BYTES_PER_CHUNK) + 1000
-        self._assert_get_chunk_specs(size, 205, s3_util.MULTIPART_UPLOAD_THRESHOLD)
-
-        size *= 2
-        self._assert_get_chunk_specs(size, 410, s3_util.MULTIPART_UPLOAD_THRESHOLD)
-
-    def test_generate_even_chunks(self):
-        generator = self.client._generate_chunks(1000, 4, 250)
-        output = list(generator)
-        expected_output = [(1, 0, 250), (2, 250, 250), (3, 500, 250), (4, 750, 250)]
-        self.assertEquals(output, expected_output)
-
-    def test_generate_uneven_chunks(self):
-        generator = self.client._generate_chunks(900, 4, 250)
-        output = list(generator)
-        expected_output = [(1, 0, 250), (2, 250, 250), (3, 500, 250), (4, 750, 150)]
-        self.assertEquals(output, expected_output)
diff --git a/edx/analytics/tasks/util/tests/test_url.py b/edx/analytics/tasks/util/tests/test_url.py
index 92a68b9e84..a84ba4c83a 100644
--- a/edx/analytics/tasks/util/tests/test_url.py
+++ b/edx/analytics/tasks/util/tests/test_url.py
@@ -3,9 +3,8 @@
 from unittest import TestCase
 
 import luigi
-import luigi.format
-import luigi.hdfs
-import luigi.s3
+from luigi.contrib.hdfs.target import HdfsTarget
+from luigi.contrib.s3 import S3Target
 from mock import patch
 
 from edx.analytics.tasks.util import url
@@ -17,12 +16,12 @@ class TargetFromUrlTestCase(TestCase):
     def test_s3_scheme(self):
         for test_url in ['s3://foo/bar', 's3n://foo/bar']:
             target = url.get_target_from_url(test_url)
-            self.assertIsInstance(target, luigi.hdfs.HdfsTarget)
+            self.assertIsInstance(target, HdfsTarget)
             self.assertEquals(target.path, test_url)
 
     def test_hdfs_scheme(self):
         target = url.get_target_from_url('hdfs:///foo/bar')
-        self.assertIsInstance(target, luigi.hdfs.HdfsTarget)
+        self.assertIsInstance(target, HdfsTarget)
         self.assertEquals(target.path, '/foo/bar')
 
     def test_file_scheme(self):
@@ -32,19 +31,20 @@ def test_file_scheme(self):
             self.assertIsInstance(target, luigi.LocalTarget)
             self.assertEquals(target.path, path)
 
-    @patch('edx.analytics.tasks.util.s3_util.connect_s3')
-    def test_s3_https_scheme(self, _mock_boto):
+    @patch('luigi.contrib.s3.S3Client')
+    def test_s3_https_scheme(self, _mock_client):
         test_url = 's3+https://foo/bar'
         target = url.get_target_from_url(test_url)
-        self.assertIsInstance(target, luigi.s3.S3Target)
+        self.assertIsInstance(target, S3Target)
         self.assertEquals(target.path, test_url)
 
     def test_hdfs_directory(self):
         test_url = 's3://foo/bar/'
         target = url.get_target_from_url(test_url)
-        self.assertIsInstance(target, luigi.hdfs.HdfsTarget)
+        self.assertIsInstance(target, HdfsTarget)
         self.assertEquals(target.path, test_url[:-1])
-        self.assertEquals(target.format, luigi.hdfs.PlainDir)
+        # TODO: target.format is wrapped.  Unwrap it....
+        # self.assertEquals(target.format, luigi.hdfs.PlainDir)
 
 
 class UrlPathJoinTestCase(TestCase):
diff --git a/edx/analytics/tasks/util/url.py b/edx/analytics/tasks/util/url.py
index dbf01be7e7..78c1932a93 100644
--- a/edx/analytics/tasks/util/url.py
+++ b/edx/analytics/tasks/util/url.py
@@ -17,13 +17,13 @@
 
 import luigi
 import luigi.configuration
-import luigi.format
-import luigi.hdfs
-import luigi.s3
-from luigi.hdfs import HdfsTarget
-from luigi.s3 import S3Target
+import luigi.contrib.hdfs
+import luigi.contrib.s3
+from luigi.contrib.hdfs import format as hdfs_format
+from luigi.contrib.hdfs.target import HdfsTarget
+from luigi.contrib.s3 import S3Target
 
-from edx.analytics.tasks.util.s3_util import S3HdfsTarget, ScalableS3Client
+from edx.analytics.tasks.util.s3_util import DEFAULT_KEY_ACCESS_POLICY, S3HdfsTarget, ScalableS3Client
 
 log = logging.getLogger(__name__)
 
@@ -90,7 +90,7 @@ def complete(self):
         return True
 
 
-class IgnoredTarget(luigi.hdfs.HdfsTarget):
+class IgnoredTarget(HdfsTarget):
     """Dummy target for use in Hadoop jobs that produce no explicit output file."""
     def __init__(self):
         super(IgnoredTarget, self).__init__(is_tmp=True)
@@ -104,11 +104,11 @@ def open(self, mode='r'):
 
 DEFAULT_TARGET_CLASS = luigi.LocalTarget
 URL_SCHEME_TO_TARGET_CLASS = {
-    'hdfs': luigi.hdfs.HdfsTarget,
+    'hdfs': HdfsTarget,
     's3': S3HdfsTarget,
     's3n': S3HdfsTarget,
     'file': luigi.LocalTarget,
-    's3+https': luigi.s3.S3Target,
+    's3+https': S3Target,
 }
 
 DEFAULT_MARKER_TARGET_CLASS = LocalMarkerTarget
@@ -131,14 +131,15 @@ def get_target_class_from_url(url, marker=False):
         target_class = URL_SCHEME_TO_TARGET_CLASS.get(parsed_url.scheme, DEFAULT_TARGET_CLASS)
 
     kwargs = {}
-    if issubclass(target_class, luigi.hdfs.HdfsTarget) and url.endswith('/'):
-        kwargs['format'] = luigi.hdfs.PlainDir
+    if issubclass(target_class, HdfsTarget) and url.endswith('/'):
+        kwargs['format'] = hdfs_format.PlainDir
     if issubclass(target_class, luigi.LocalTarget) or parsed_url.scheme == 'hdfs':
         # LocalTarget and HdfsTarget both expect paths without any scheme, netloc etc, just bare paths. So strip
         # everything else off the url and pass that in to the target.
         url = parsed_url.path
-    if issubclass(target_class, luigi.s3.S3Target):
+    if issubclass(target_class, S3Target):
         kwargs['client'] = ScalableS3Client()
+        kwargs['policy'] = DEFAULT_KEY_ACCESS_POLICY
 
     url = url.rstrip('/')
     args = (url,)
diff --git a/edx/analytics/tasks/warehouse/financial/cybersource.py b/edx/analytics/tasks/warehouse/financial/cybersource.py
index 1d88a89781..dd37bf0f2e 100644
--- a/edx/analytics/tasks/warehouse/financial/cybersource.py
+++ b/edx/analytics/tasks/warehouse/financial/cybersource.py
@@ -260,9 +260,8 @@ class CybersourceDataValidationTask(WarehouseMixin, luigi.WrapperTask):
 
     import_date = luigi.DateParameter()
 
-    cybersource_merchant_ids = luigi.Parameter(
-        default_from_config={'section': 'payment', 'name': 'cybersource_merchant_ids'},
-        is_list=True
+    cybersource_merchant_ids = luigi.ListParameter(
+        config_path={'section': 'payment', 'name': 'cybersource_merchant_ids'},
     )
 
     def requires(self):
diff --git a/edx/analytics/tasks/warehouse/financial/finance_reports.py b/edx/analytics/tasks/warehouse/financial/finance_reports.py
index dfae8c14d5..e26380e314 100644
--- a/edx/analytics/tasks/warehouse/financial/finance_reports.py
+++ b/edx/analytics/tasks/warehouse/financial/finance_reports.py
@@ -21,7 +21,7 @@ class BuildFinancialReportsTask(MapReduceJobTaskMixin, VerticaCopyTaskMixin, lui
 
     # Redefine the overwrite parameter to change its default to True.
     # This will cause the reports to reload when loading into internal reporting.
-    overwrite = luigi.BooleanParameter(default=True)
+    overwrite = luigi.BoolParameter(default=True)
 
     def requires(self):
         yield (
diff --git a/edx/analytics/tasks/warehouse/financial/orders_import.py b/edx/analytics/tasks/warehouse/financial/orders_import.py
index 072b7fe3d7..f639c856a1 100644
--- a/edx/analytics/tasks/warehouse/financial/orders_import.py
+++ b/edx/analytics/tasks/warehouse/financial/orders_import.py
@@ -1,6 +1,6 @@
 """Import Orders: Shopping Cart Tables from the LMS, Orders from Otto."""
 
-import luigi.hdfs
+import luigi
 
 from edx.analytics.tasks.insights.database_imports import (
     DatabaseImportMixin, ImportAuthUserTask, ImportCouponVoucherIndirectionState, ImportCouponVoucherState,
@@ -16,10 +16,10 @@
 
 class OrderTableTask(DatabaseImportMixin, HiveTableFromQueryTask):
     otto_credentials = luigi.Parameter(
-        default_from_config={'section': 'otto-database-import', 'name': 'credentials'}
+        config_path={'section': 'otto-database-import', 'name': 'credentials'}
     )
     otto_database = luigi.Parameter(
-        default_from_config={'section': 'otto-database-import', 'name': 'database'}
+        config_path={'section': 'otto-database-import', 'name': 'database'}
     )
 
     def requires(self):
diff --git a/edx/analytics/tasks/warehouse/financial/payment.py b/edx/analytics/tasks/warehouse/financial/payment.py
index 3492c06e4f..1ac0d1bb71 100644
--- a/edx/analytics/tasks/warehouse/financial/payment.py
+++ b/edx/analytics/tasks/warehouse/financial/payment.py
@@ -10,9 +10,8 @@
 class PaymentTask(luigi.WrapperTask):
 
     import_date = luigi.DateParameter()
-    cybersource_merchant_ids = luigi.Parameter(
-        default_from_config={'section': 'payment', 'name': 'cybersource_merchant_ids'},
-        is_list=True
+    cybersource_merchant_ids = luigi.ListParameter(
+        config_path={'section': 'payment', 'name': 'cybersource_merchant_ids'},
     )
 
     def requires(self):
diff --git a/edx/analytics/tasks/warehouse/financial/paypal.py b/edx/analytics/tasks/warehouse/financial/paypal.py
index abe6d98661..f88f674fd7 100644
--- a/edx/analytics/tasks/warehouse/financial/paypal.py
+++ b/edx/analytics/tasks/warehouse/financial/paypal.py
@@ -571,7 +571,7 @@ class PaypalTaskMixin(OverwriteOutputMixin):
         description='The date to generate a report for. Default is today, UTC.',
     )
     account_id = luigi.Parameter(
-        default_from_config={'section': 'paypal', 'name': 'account_id'},
+        config_path={'section': 'paypal', 'name': 'account_id'},
         description='A human readable name for the paypal account data is being gathered for.',
     )
 
@@ -678,7 +678,7 @@ class PaypalTransactionsIntervalTask(PaypalTaskMixin, WarehouseMixin, luigi.Wrap
 
     date = None
     interval_start = luigi.DateParameter(
-        default_from_config={'section': 'paypal', 'name': 'interval_start'},
+        config_path={'section': 'paypal', 'name': 'interval_start'},
         significant=False,
     )
     interval_end = luigi.DateParameter(
@@ -736,7 +736,7 @@ class PaypalDataValidationTask(WarehouseMixin, luigi.WrapperTask):
     import_date = luigi.DateParameter()
 
     paypal_interval_start = luigi.DateParameter(
-        default_from_config={'section': 'paypal', 'name': 'interval_start'},
+        config_path={'section': 'paypal', 'name': 'interval_start'},
         significant=False,
     )
 
diff --git a/edx/analytics/tasks/warehouse/load_internal_reporting_course_catalog.py b/edx/analytics/tasks/warehouse/load_internal_reporting_course_catalog.py
index 90df883ccd..7ed05155ff 100644
--- a/edx/analytics/tasks/warehouse/load_internal_reporting_course_catalog.py
+++ b/edx/analytics/tasks/warehouse/load_internal_reporting_course_catalog.py
@@ -7,7 +7,7 @@
 import logging
 
 import luigi
-from luigi.hive import HiveQueryTask
+from luigi.contrib.hive import HiveQueryTask
 
 from edx.analytics.tasks.common.vertica_load import VerticaCopyTask, VerticaCopyTaskMixin
 from edx.analytics.tasks.util.edx_api_client import EdxApiClient
@@ -31,16 +31,14 @@ class LoadInternalReportingCourseCatalogMixin(WarehouseMixin, OverwriteOutputMix
         default=datetime.datetime.utcnow().date(),
         description='Default is today, UTC.',
     )
-    partner_short_codes = luigi.Parameter(
-        default_from_config={'section': 'course-catalog-api', 'name': 'partner_short_codes'},
-        is_list=True,
-        default=None,
+    partner_short_codes = luigi.ListParameter(
+        config_path={'section': 'course-catalog-api', 'name': 'partner_short_codes'},
+        default=list(),
         description="A list of partner short codes that we should fetch data for."
     )
-    partner_api_urls = luigi.Parameter(
-        default_from_config={'section': 'course-catalog-api', 'name': 'partner_api_urls'},
-        is_list=True,
-        default=None,
+    partner_api_urls = luigi.ListParameter(
+        config_path={'section': 'course-catalog-api', 'name': 'partner_api_urls'},
+        default=list(),
         description="A list of API URLs that are associated with the partner_short_codes.  This list must exactly " +
                     "match the ordering of the partner_short_codes list."
     )
diff --git a/edx/analytics/tasks/warehouse/load_internal_reporting_database.py b/edx/analytics/tasks/warehouse/load_internal_reporting_database.py
index 055091c00d..4b1c63abb5 100644
--- a/edx/analytics/tasks/warehouse/load_internal_reporting_database.py
+++ b/edx/analytics/tasks/warehouse/load_internal_reporting_database.py
@@ -138,7 +138,6 @@ class PreImportDatabaseTask(SchemaManagementTask):
     """
     Task needed to run before importing database into warehouse.
     """
-    priority = 100
 
     @property
     def queries(self):
@@ -159,12 +158,10 @@ class PostImportDatabaseTask(SchemaManagementTask):
     """
     Task needed to run after importing database into warehouse.
     """
-    priority = -100
 
     # Override the standard roles here since these tables will be rather raw. We may want to restrict access to a
     # subset of users.
-    roles = luigi.Parameter(
-        is_list=True,
+    roles = luigi.ListParameter(
         config_path={'section': 'vertica-export', 'name': 'business_intelligence_team_roles'},
     )
 
@@ -205,13 +202,12 @@ class ImportMysqlToVerticaTask(MysqlToVerticaTaskMixin, luigi.WrapperTask):
     date = luigi.DateParameter(
         default=datetime.datetime.utcnow().date(),
     )
-    overwrite = luigi.BooleanParameter(
+    overwrite = luigi.BoolParameter(
         default=False,
         significant=False,
     )
 
-    exclude = luigi.Parameter(
-        is_list=True,
+    exclude = luigi.ListParameter(
         default=(),
     )
 
@@ -222,14 +218,17 @@ class ImportMysqlToVerticaTask(MysqlToVerticaTaskMixin, luigi.WrapperTask):
     def __init__(self, *args, **kwargs):
         super(ImportMysqlToVerticaTask, self).__init__(*args, **kwargs)
         self.table_list = []
+        self.is_complete = False
 
     def should_exclude_table(self, table_name):
-        """Determines whether to exlude a table during the import."""
+        """Determines whether to exclude a table during the import."""
         if any(re.match(pattern, table_name) for pattern in self.exclude):
             return True
         return False
 
-    def requires(self):
+    def run(self):
+        # Add yields of tasks in run() method, to serve as dynamic dependencies.
+        # This method should be rerun each time it yields a job.
         if not self.table_list:
             results = get_mysql_query_results(self.db_credentials, self.database, 'show tables')
             self.table_list = [result[0].strip() for result in results]
@@ -239,7 +238,7 @@ def requires(self):
             schema=self.schema,
             credentials=self.credentials,
             marker_schema=self.marker_schema,
-            overwrite=self.overwrite
+            overwrite=self.overwrite,
         )
         yield pre_import_task
 
@@ -262,5 +261,9 @@ def requires(self):
             schema=self.schema,
             credentials=self.credentials,
             marker_schema=self.marker_schema,
-            overwrite=self.overwrite
+            overwrite=self.overwrite,
         )
+        self.is_complete = True
+
+    def complete(self):
+        return self.is_complete
diff --git a/edx/analytics/tasks/warehouse/load_internal_reporting_events.py b/edx/analytics/tasks/warehouse/load_internal_reporting_events.py
index 7ceafb7119..05b17c6dde 100644
--- a/edx/analytics/tasks/warehouse/load_internal_reporting_events.py
+++ b/edx/analytics/tasks/warehouse/load_internal_reporting_events.py
@@ -933,13 +933,11 @@ def mapper(self, line):
 class SegmentEventLogSelectionDownstreamMixin(EventLogSelectionDownstreamMixin):
     """Defines parameters for passing upstream to tasks that use SegmentEventLogSelectionMixin."""
 
-    source = luigi.Parameter(
-        is_list=True,
+    source = luigi.ListParameter(
         config_path={'section': 'segment-logs', 'name': 'source'},
         description='A URL to a path that contains log files that contain the events. (e.g., s3://my_bucket/foo/).   Segment-logs',
     )
-    pattern = luigi.Parameter(
-        is_list=True,
+    pattern = luigi.ListParameter(
         config_path={'section': 'segment-logs', 'name': 'pattern'},
         description='A regex with a named capture group for the date or timestamp that approximates the date that the events '
         'within were emitted. Note that the search interval is expanded, so events don\'t have to be in exactly '
@@ -1555,8 +1553,7 @@ class PruneEventPartitionsInVertica(EventRecordLoadDownstreamMixin, SchemaManage
 
     # Override the standard roles here since these tables will be rather raw. We may want to restrict access to a
     # subset of users.
-    roles = luigi.Parameter(
-        is_list=True,
+    roles = luigi.ListParameter(
         config_path={'section': 'vertica-export', 'name': 'restricted_roles'},
     )
 
diff --git a/edx/analytics/tasks/warehouse/load_internal_reporting_user_activity.py b/edx/analytics/tasks/warehouse/load_internal_reporting_user_activity.py
index 95a2f21c85..7162920436 100644
--- a/edx/analytics/tasks/warehouse/load_internal_reporting_user_activity.py
+++ b/edx/analytics/tasks/warehouse/load_internal_reporting_user_activity.py
@@ -268,13 +268,13 @@ def requires(self):
 
 class UserActivityWorkflow(WeeklyIntervalMixin, WarehouseMixin, luigi.WrapperTask):
 
-    overwrite_hive = luigi.BooleanParameter(
+    overwrite_hive = luigi.BoolParameter(
         default=False,
         description='Whether or not to overwrite Hive data.',
         significant=False
     )
 
-    overwrite_mysql = luigi.BooleanParameter(
+    overwrite_mysql = luigi.BoolParameter(
         default=False,
         description='Whether or not to overwrite existing outputs in Mysql.',
         significant=False
diff --git a/edx/analytics/tasks/warehouse/load_warehouse.py b/edx/analytics/tasks/warehouse/load_warehouse.py
index eaccadc5db..121ef0f633 100644
--- a/edx/analytics/tasks/warehouse/load_warehouse.py
+++ b/edx/analytics/tasks/warehouse/load_warehouse.py
@@ -43,7 +43,7 @@ class WarehouseWorkflowMixin(WarehouseMixin):
         description='Path to the external access credentials file.',
     )
 
-    overwrite = luigi.BooleanParameter(default=False, significant=False)
+    overwrite = luigi.BoolParameter(default=False, significant=False)
 
     # We rename the schema after warehouse loading step. This causes
     # Luigi to think that tasks are not complete as it cannot find the
@@ -56,7 +56,6 @@ class PreLoadWarehouseTask(SchemaManagementTask):
     """
     Task needed to run before loading data into warehouse.
     """
-    priority = 100
 
     @property
     def queries(self):
@@ -75,14 +74,6 @@ class LoadWarehouseTask(WarehouseWorkflowMixin, luigi.WrapperTask):
     """Runs the tasks needed to load warehouse."""
 
     def requires(self):
-        kwargs = {
-            'schema': self.schema + '_loading',
-            'marker_schema': self.marker_schema,
-            'credentials': self.credentials,
-            'overwrite': self.overwrite,
-            'warehouse_path': self.warehouse_path,
-        }
-
         yield PreLoadWarehouseTask(
             date=self.date,
             schema=self.schema,
@@ -90,7 +81,19 @@ def requires(self):
             marker_schema=self.marker_schema,
             overwrite=self.overwrite
         )
-        yield (
+
+    def __init__(self, *args, **kwargs):
+        super(LoadWarehouseTask, self).__init__(*args, **kwargs)
+        self.is_complete = False
+
+        kwargs = {
+            'schema': self.schema + '_loading',
+            'marker_schema': self.marker_schema,
+            'credentials': self.credentials,
+            'overwrite': self.overwrite,
+            'warehouse_path': self.warehouse_path,
+        }
+        self.tasks_to_run = [
             LoadInternalReportingCertificatesToWarehouse(
                 date=self.date,
                 **kwargs
@@ -118,7 +121,18 @@ def requires(self):
                 n_reduce_tasks=self.n_reduce_tasks,
                 **kwargs
             ),
-        )
+        ]
+
+    def run(self):
+        # This is called again after each yield that provides a dynamic dependency.
+        # Because overwrite might be set on the tasks, make sure that each is called only once.
+        if self.tasks_to_run:
+            yield self.tasks_to_run.pop()
+
+        self.is_complete = True
+
+    def complete(self):
+        return self.is_complete
 
 
 class PostLoadWarehouseTask(SchemaManagementTask):
@@ -127,8 +141,6 @@ class PostLoadWarehouseTask(SchemaManagementTask):
     """
     n_reduce_tasks = luigi.Parameter()
 
-    priority = -100
-
     def requires(self):
         return {
             'source': LoadWarehouseTask(
diff --git a/requirements/extra.txt b/requirements/extra.txt
index b3ea7306f0..1b281bfa08 100644
--- a/requirements/extra.txt
+++ b/requirements/extra.txt
@@ -4,7 +4,8 @@
 http://cdn.mysql.com/Downloads/Connector-Python/mysql-connector-python-1.2.2.zip  	# GPL v2 with FOSS License Exception
 
 # Putting these here rather than in default.in allows us to avoid using editable mode that pip-compile requires.
-git+https://github.com/edx/luigi.git@a73700ca51685974220ef6069d2f078312055444#egg=luigi 		# Apache License 2.0
+git+https://github.com/edx/luigi.git@5472d77309b4600de9e8c246935d17972d0a6c9f#egg=luigi # Apache License 2.0
+
 git+https://github.com/edx/pyinstrument.git@a35ff76df4c3d5ff9a2876d859303e33d895e78f#egg=pyinstrument     # BSD
 
 # This was originally separated out from default.in so that it was not included in docs.txt.  Not clear why.
diff --git a/setup.cfg b/setup.cfg
index 16d804fbff..49c5e67380 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -95,7 +95,7 @@ edx.analytics.tasks =
 
 mapreduce.engine =
     hadoop = edx.analytics.tasks.common.mapreduce:MapReduceJobRunner
-    local = luigi.hadoop:LocalJobRunner
+    local = luigi.contrib.hadoop:LocalJobRunner
     emu = edx.analytics.tasks.common.mapreduce:EmulatedMapReduceJobRunner
 
 [pycodestyle]