Add configure_job method to BigQuery job tasks

luigi/contrib/bigquery.py

@@ -527,6 +527,16 @@ def allow_quoted_new_lines(self):
         """	Indicates if BigQuery should allow quoted data sections that contain newline characters in a CSV file. The default value is false."""
         return False
 
+    def configure_job(self, configuration):
+        """Set additional job configuration.
+
+        This allows to specify job configuration parameters that are not exposed via Task properties.
+
+        :param configuration: Current configuration.
+        :return: New or updated configuration.
+        """
+        return configuration
+
     def run(self):
         output = self.output()
         assert isinstance(output, BigQueryTarget), 'Output must be a BigQueryTarget, not %s' % (output)
@@ -565,6 +575,8 @@ def run(self):
         else:
             job['configuration']['load']['autodetect'] = True
 
+        job['configuration'] = self.configure_job(job['configuration'])
+
         bq_client.run_job(output.table.project_id, job, dataset=output.table.dataset)
 
 
@@ -610,6 +622,16 @@ def use_legacy_sql(self):
         """
         return True
 
+    def configure_job(self, configuration):
+        """Set additional job configuration.
+
+        This allows to specify job configuration parameters that are not exposed via Task properties.
+
+        :param configuration: Current configuration.
+        :return: New or updated configuration.
+        """
+        return configuration
+
     def run(self):
         output = self.output()
         assert isinstance(output, BigQueryTarget), 'Output must be a BigQueryTarget, not %s' % (output)
@@ -643,6 +665,8 @@ def run(self):
             }
         }
 
+        job['configuration'] = self.configure_job(job['configuration'])
+
         bq_client.run_job(output.table.project_id, job, dataset=output.table.dataset)
 
 
@@ -739,6 +763,16 @@ def compression(self):
         """Whether to use compression."""
         return Compression.NONE
 
+    def configure_job(self, configuration):
+        """Set additional job configuration.
+
+        This allows to specify job configuration parameters that are not exposed via Task properties.
+
+        :param configuration: Current configuration.
+        :return: New or updated configuration.
+        """
+        return configuration
+
     def run(self):
         input = luigi.task.flatten(self.input())[0]
         assert (
@@ -775,6 +809,8 @@ def run(self):
             job['configuration']['extract']['fieldDelimiter'] = \
                 self.field_delimiter
 
+        job['configuration'] = self.configure_job(job['configuration'])
+
         bq_client.run_job(
             input.table.project_id,
             job,