[GitHub] [airflow] Orpheuz commented on a change in pull request #4460: [AIRFLOW-3604] Add Azure Hdinsight operators for creating cluster and submitting

[GitBox <gi...@apache.org>]

Orpheuz commented on a change in pull request #4460: [AIRFLOW-3604] Add Azure Hdinsight operators for creating cluster and submitting
URL: https://github.com/apache/airflow/pull/4460#discussion_r272111495
 
 

 ##########
 File path: airflow/contrib/hooks/hortonworks_ambari_hook.py
 ##########
 @@ -0,0 +1,131 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# 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
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+import json
+import time
+
+from airflow.hooks.http_hook import HttpHook
+from airflow.exceptions import AirflowException
+
+
+class HdpAmbariHook(HttpHook):
+
+    def __init__(self, ambari_conn_id='hortonworks_ambari_default'):
+        """
+
+        :param azure_conn_id: connection id of Hortonworks Ambari cluster
+                    which will be used to Hortonworks rest service
+        :type ambari_conn_id: str
+        """
+        super(HdpAmbariHook, self).__init__(ambari_conn_id)
+
+        self.headers = {
+            'Content-Type': "application/x-www-form-urlencoded"
+        }
+        connection = self.get_connection(ambari_conn_id)
+        self.cluster_name = str(connection.host).split("//")[0].split(".")[0]  # clustername
+        username = connection.login
+        self.params = {"user.name": username}
+
+    def submit_hive_job(self, datas: dict):
+        """
+        Executes hql code or hive script in Azure HDInsight Cluster
+
+        :param datas: parametres of Hive script
+        :type datas: object
+
+        """
+
+        self.method = "POST"
+        hive_endpoint = "templeton/v1/"
+        statement_non_terminated_status_list = ['INIT', 'RUNNING', 'PREP']
+        submit_endpoint = hive_endpoint + "hive"
+
+        self.log.debug("Submiting hive  Script: %s", str(datas))
+        response = self.run(endpoint=submit_endpoint, data=datas, headers=self.headers)
+
+        job_id = response["id"]
+        status_endpoint = hive_endpoint + "jobs/" + str(job_id)
+        self.log.info("Finished submitting hive script job_id: %s", job_id)
+
+        statements_state = 'INIT'
+        while statements_state in statement_non_terminated_status_list:
+
+            # todo: test execution_timeout
+            time.sleep(5)
+            try:
+                result = self._get_job_status(endpoint=status_endpoint)["status"]["state"]
+            except Exception:
+                pass
+
+            statements_state = result["status"]["state"]
+            statements_exit_value = result["exitValue"]
+            if statements_state is not None:
+                # Check submitted job's result
+                if statements_state in ['KILLED', 'FAILED'] or statements_exit_value != "0":
+                    result = "job failed. state: %s", statements_state
+                    self.log.error(result)
+                    raise AirflowException(result)
+                else:
+                    self.log.debug("Checking Hive job: %s", statements_state)
+
+        self.log.debug("Statement %s ", statements_state)
+        self.log.debug("Finished executing WebHCatHiveSubmitOperator")
+
+    def submit_spark_job(self, datas):
+        """
+        Submit spark job or spark-sql to Azure HDInsight Cluster
+
+        :param datas: parametres of Spark Job
+        :type datas: object
+        """
+
+        statement_non_terminated_status_list = ['INIT', 'starting', 'running', "waiting", "available"]
+        spark_endpoint = "livy/batches"
+        submit_endpoint = spark_endpoint
+        self.method = "POST"
+        datas["user.name"] = self.username
+
+        self.log.info("Submiting spark Job: %s ", json.dumps(datas))
+        response = self.run(endpoint=submit_endpoint, data=json.dumps(datas))
 
 Review comment:
   Aren't headers missing here? They might be necessary in order to satisfy Ambary CSRF checks

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services