def __init__(self, *args, **kwargs):
     conn = self.get_connection(kwargs['qubole_conn_id'])
     Qubole.configure(api_token=conn.password, api_url=conn.host)
     self.task_id = kwargs['task_id']
     self.dag_id = kwargs['dag'].dag_id
     self.kwargs = kwargs
     self.cls = COMMAND_CLASSES[self.kwargs['command_type']]
     self.cmd = None

 def _configure_qubole(self):
     logging.basicConfig(level=logging.INFO)
     logger = logging.getLogger('qds_connection')
     logger.propagate = False
     qdslog = logging.getLogger('qds')
     if not self.config.API_TOKEN:
         raise Exception("You didn't specify your QUBOLE_API_TOKEN in "
                         "your environment before running commands on "
                         "Qubole!\n. It can be found at http://api.qubole"
                         ".com/users/edit")
     Qubole.configure(api_token=self.config.API_TOKEN,
                      api_url=self.config.API_URL,
                      version=self.config.API_VERSION,
                      poll_interval=self.config.POLL_INTERVAL_SEC)
     return qdslog

 def update(cls, cluster_id_label, cluster_info):
     """
     Update the cluster with id/label `cluster_id_label` using information provided in
     `cluster_info`.
     """
     conn = Qubole.agent(version="v2")
     return conn.put(cls.element_path(cluster_id_label), data=cluster_info)

 def terminate(cls, cluster_id_label):
     """
     Terminate the cluster with id/label `cluster_id_label`.
     """
     conn = Qubole.agent()
     data = {"state": "terminate"}
     return conn.put(cls.element_path(cluster_id_label) + "/state", data)

 def clone(cls, cluster_id_label, cluster_info):
     """
     Update the cluster with id/label `cluster_id_label` using information provided in
     `cluster_info`.
     """
     conn = Qubole.agent()
     return conn.post(cls.element_path(cluster_id_label) + '/clone', data=cluster_info)

 def find_by_name(name):
     conn = Qubole.agent()
     if name is not None:
         schedjson = conn.get(Scheduler.rest_entity_path, params={"name":name})
         if schedjson["schedules"]:
             return Scheduler(schedjson["schedules"][0])
     return None

    def poke(self, context):
        conn = BaseHook.get_connection(self.qubole_conn_id)
        Qubole.configure(api_token=conn.password, api_url=conn.host)

        this.log.info('Poking: %s', self.data)

        status = False
        try:
            status = self.sensor_class.check(self.data)
        except Exception as e:
            logging.exception(e)
            status = False

        this.log.info('Status of this Poke: %s', status)

        return status

 def add_node(cls, cluster_id_label, parameters=None):
   """
   Add a node to an existing cluster
   """
   conn = Qubole.agent()
   parameters = {} if not parameters else parameters
   return conn.post(cls.element_path(cluster_id_label) + "/nodes", data={"parameters" : parameters})

 def index(cls):
     """
     Shows a list of all available reports by issuing a GET request to the
     /reports endpoint.
     """
     conn = Qubole.agent()
     return conn.get(cls.rest_entity_path)

 def start(cls, cluster_id_label):
     """
     Start the cluster with id/label `cluster_id_label`.
     """
     conn = Qubole.agent()
     data = {"state": "start"}
     return conn.put(cls.element_path(cluster_id_label) + "/state", data)

 def find(cls, name="default", **kwargs):
     if (name is None) or (name == "default"):
         conn = Qubole.agent()
         return cls(conn.get(cls.rest_entity_path))
     else:
         raise ParseError("Bad name %s" % name,
                          "Hadoop Clusters can only be named 'default' currently")

    def get_results(self, fp=sys.stdout, inline=True, delim=None):
        """
        Fetches the result for the command represented by this object

        Args:
            `fp`: a file object to write the results to directly
        """
        result_path = self.meta_data['results_resource']

        conn = Qubole.agent()

        r = conn.get(result_path, {'inline': inline})
        if r.get('inline'):
            if sys.version_info < (3, 0, 0):
                fp.write(r['results'].encode('utf8'))
            else:
                import io
                if isinstance(fp, io.TextIOBase):
                    fp.buffer.write(r['results'].encode('utf8'))
                elif isinstance(fp, io.BufferedIOBase) or isinstance(fp, io.RawIOBase):
                    fp.write(r['results'].encode('utf8'))
                else:
                    # Can this happen? Don't know what's the right thing to do in this case.
                    pass
        else:
            acc = Account.find()
            boto_conn = boto.connect_s3(aws_access_key_id=acc.storage_access_key,
                                        aws_secret_access_key=acc.storage_secret_key)

            log.info("Starting download from result locations: [%s]" % ",".join(r['result_location']))
            #fetch latest value of num_result_dir
            num_result_dir = Command.find(self.id).num_result_dir
            for s3_path in r['result_location']:
                # In Python 3, in this case, `fp` should always be binary mode.
                _download_to_local(boto_conn, s3_path, fp, num_result_dir, delim=delim)

    def list(cls, label=None, cluster_id=None, state=None):
        """
        List existing clusters present in your account.

        Kwargs:
            `state`: list only those clusters which are in this state

        Returns:
            List of clusters satisfying the given criteria
        """
        if cluster_id is not None:
            return cls.show(cluster_id)
        if label is not None:
            return cls.show(label)
        conn = Qubole.agent(version="v2")
        cluster_list = conn.get(cls.rest_entity_path)
        if state is None:
            # return the complete list since state is None
            return conn.get(cls.rest_entity_path)
        # filter clusters based on state
        result = []
        if 'clusters' in cluster_list:
            for cluster in cluster_list['clusters']:
                if state.lower() == cluster['state'].lower():
                    result.append(cluster)
        return result

 def update_node(cls, cluster_id_label, command, private_dns, parameters=None):
     """
     Add a node to an existing cluster
     """
     conn = Qubole.agent()
     parameters = {} if not parameters else parameters
     data = {"command" : command, "private_dns" : private_dns, "parameters" : parameters}
     return conn.put(cls.element_path(cluster_id_label) + "/nodes", data)

 def check(cls, data):
     """
     Method to call the sensors api with json payload
     :param data: valid json object
     :return: True or False
     """
     conn = Qubole.agent()
     return conn.post(cls.rest_entity_path, data=data)['status']

 def __init__(self, access=None, secret = None, testmode=False, db_parallelism=None, mode=None, db_table=None, db_where=None, db_columns=None, db_boundary_query=None, db_extract_query=None, db_split_column=None, hive_table=None, part_spec=None, db_user=None, db_passwd=None, db_host=None, db_port=None, db_type=None, db_name=None, api_token = None, api_url=None, fetch_size = None):
   self.temp_location = "/tmp/sqoop/"+uuid.uuid1().hex
   self.tmp_dir = tempfile.mkdtemp(prefix="/media/ephemeral0/logs"+"/sqoop")
   logger.info("Temp Directory is:" + self.tmp_dir)
   self.access = access
   self.secret = secret
   self.api_token = api_token
   self.api_url = api_url
   self.fetch_size = fetch_size
   self.redshift_sink = False
   self.__loadImportParamsFromCid(testmode, db_parallelism, mode, db_table, db_where, db_columns, db_boundary_query, db_extract_query, db_split_column, hive_table, part_spec, db_user, db_passwd, db_host, db_port, db_type, db_name)
   self.sqoop_cmd=["/usr/lib/sqoop-h2/bin/sqoop"]
   self.sqoop_cmd.extend(["import"])
   self.__addBasicOptions()
   self.__extendCmdSpecificOptions()
   Qubole.configure(api_token=api_token, api_url=api_url)
   self.cluster_label = Cluster.show(os.popen("cat /usr/lib/hustler/bin/nodeinfo_src.sh | grep cluster_id").read().split("=")[1].strip().replace('"',''))['cluster']['label'][0]

    def create_update_clone_parser(subparser, action=None):
        # cloud config parser
        cloud = Qubole.get_cloud()
        cloud.create_parser(subparser)

        # cluster info parser
        ClusterInfoV2.cluster_info_parser(subparser, action)

        # engine config parser
        Engine.engine_parser(subparser)

    def cancel_id(cls, id):
        """
        Cancels command denoted by this id

        Args:
            `id`: command id
        """
        conn = Qubole.agent()
        data = {"status": "kill"}
        return conn.put(cls.element_path(id), data)

    def get_log_id(cls, id):
        """
        Fetches log for the command represented by this id

        Args:
            `id`: command id
        """
        conn = Qubole.agent()
        r = conn.get_raw(cls.element_path(id) + "/logs")
        return r.text

 def snapshot(cls, cluster_id_label, s3_location, backup_type):
     """
     Create hbase snapshot full/incremental
     """
     conn = Qubole.agent()
     parameters = {}
     parameters['s3_location'] = s3_location
     if backup_type:
         parameters['backup_type'] = backup_type
     return conn.post(cls.element_path(cluster_id_label) + "/snapshots", data=parameters)