Repository: incubator-airflow Updated Branches: refs/heads/master 6b2fdbef0 -> 2f009164d
[AIRFLOW-2751] add job properties update in hive to druid operator. Closes #3600 from happyjulie/AIRFLOW-2751 Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/2f009164 Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/2f009164 Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/2f009164 Branch: refs/heads/master Commit: 2f009164df4f45291000cb991c7bd8ff99b479da Parents: 6b2fdbe Author: juhwi.lee <[email protected]> Authored: Sun Jul 15 12:14:21 2018 +0200 Committer: Fokko Driesprong <[email protected]> Committed: Sun Jul 15 12:14:21 2018 +0200 ---------------------------------------------------------------------- airflow/operators/hive_to_druid.py | 8 ++++++++ tests/operators/test_hive_to_druid.py | 17 +++++++++-------- 2 files changed, 17 insertions(+), 8 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/2f009164/airflow/operators/hive_to_druid.py ---------------------------------------------------------------------- diff --git a/airflow/operators/hive_to_druid.py b/airflow/operators/hive_to_druid.py index 9b4a881..f39f2c9 100644 --- a/airflow/operators/hive_to_druid.py +++ b/airflow/operators/hive_to_druid.py @@ -55,6 +55,8 @@ class HiveToDruidTransfer(BaseOperator): :param hive_tblproperties: additional properties for tblproperties in hive for the staging table :type hive_tblproperties: dict + :param job_properties: additional properties for job + :type job_properties: dict """ template_fields = ('sql', 'intervals') @@ -77,6 +79,7 @@ class HiveToDruidTransfer(BaseOperator): query_granularity="NONE", segment_granularity="DAY", hive_tblproperties=None, + job_properties=None, *args, **kwargs): super(HiveToDruidTransfer, self).__init__(*args, **kwargs) self.sql = sql @@ -95,6 +98,7 @@ class HiveToDruidTransfer(BaseOperator): self.druid_ingest_conn_id = druid_ingest_conn_id self.metastore_conn_id = metastore_conn_id self.hive_tblproperties = hive_tblproperties + self.job_properties = job_properties def execute(self, context): hive = HiveCliHook(hive_cli_conn_id=self.hive_cli_conn_id) @@ -231,6 +235,10 @@ class HiveToDruidTransfer(BaseOperator): } } + if self.job_properties: + ingest_query_dict['spec']['tuningConfig']['jobProperties'] \ + .update(self.job_properties) + if self.hadoop_dependency_coordinates: ingest_query_dict['hadoopDependencyCoordinates'] \ = self.hadoop_dependency_coordinates http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/2f009164/tests/operators/test_hive_to_druid.py ---------------------------------------------------------------------- diff --git a/tests/operators/test_hive_to_druid.py b/tests/operators/test_hive_to_druid.py index 06e9596..b657e29 100644 --- a/tests/operators/test_hive_to_druid.py +++ b/tests/operators/test_hive_to_druid.py @@ -7,9 +7,9 @@ # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at -# +# # http://www.apache.org/licenses/LICENSE-2.0 -# +# # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY @@ -47,7 +47,12 @@ class TestDruidHook(unittest.TestCase): 'num_shards': -1, 'target_partition_size': 1925, 'query_granularity': 'month', - 'segment_granularity': 'week' + 'segment_granularity': 'week', + 'job_properties': { + "mapreduce.job.user.classpath.first": "false", + "mapreduce.map.output.compress": "false", + "mapreduce.output.fileoutputformat.compress": "false" + } } index_spec_config = { @@ -112,11 +117,7 @@ class TestDruidHook(unittest.TestCase): }, "tuningConfig": { "type": "hadoop", - "jobProperties": { - "mapreduce.job.user.classpath.first": "false", - "mapreduce.map.output.compress": "false", - "mapreduce.output.fileoutputformat.compress": "false", - }, + "jobProperties": self.hook_config['job_properties'], "partitionsSpec": { "type": "hashed", "targetPartitionSize": self.hook_config['target_partition_size'],
