#!/usr/bin/env python
# coding=utf-8

# Copyright [2019] [Apache Software Foundation]
#
# Licensed 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.

from __future__ import print_function
import click
import time
import os
import re
import sys
import json
from paramiko import SSHClient, AutoAddPolicy
from pyhive import hive
from slugify import slugify
import hashlib

from .._logging import get_logger

from .._compatibility import six


logger = get_logger('management.hive')


@click.group('hive')
def cli():
    pass


@cli.command('hive-generateconf', help='Generate default configuration file')
@click.pass_context
def hive_generateconf_cli(ctx):
    hive_generateconf(ctx)


def hive_generateconf(ctx):
    default_conf = [{
        "origin_host": "xxx_host_name",
        "origin_db": "xxx_db_name",
        "origin_queue": "marvin",
        "target_table_name": "xxx_table_name",
        "sample_sql": "SELECT * FROM XXX",
        "sql_id": "1"
    }]

    with open('hive_dataimport.conf', 'w') as outfile:
        json.dump(default_conf, outfile, indent=2)

    print("Done!!!")


@cli.command('hive-resetremote', help='Drop all remote tables from informed engine on host')
@click.option('--host', '-h', default='marvin-hadoop')
@click.option('--queue', '-h', default='default')
@click.option('--engine', default=(os.path.relpath(".", "..")), help='Marvin engine name (default is the current folder)')
@click.pass_context
def hive_resetremote_cli(ctx, host, engine, queue):
    hive_resetremote(ctx, host, engine, queue)


def hive_resetremote(ctx, host, engine, queue):
    hdi = HiveDataImporter(
        engine=engine,
        origin_host=host,
        origin_queue=queue,
        origin_db=None,
        target_table_name=None,
        sample_sql=None,
        max_query_size=None,
        destination_host=None,
        destination_port=None,
        destination_host_username='vagrant',
        destination_host_password='vagrant',
        destination_hdfs_root_path='/user/hive/warehouse/',
        sql_id=None
    )
    hdi.reset_remote_tables()


@cli.command(
    'hive-dataimport',
    help='Export and import data samples from a hive databse to the hive running in this toolbox, cloning same data structure (db and table).')
@click.option('--destination-hdfs-root-path', '-hdfs', default='/user/hive/warehouse/')
@click.option('--destination-host-password', '-p', default='vagrant')
@click.option('--destination-host-username', '-u', default='vagrant')
@click.option('--destination-host', '-dh', default='marvin-hadoop')
@click.option('--destination-port', '-dp', default=22)
@click.option('--max-query-size', '-s', default=(50 * 1024 * 1024), help='Max query size in bytes')
@click.option('--force', is_flag=True, help='Force table creation even table already exists in destination')
@click.option('--force-remote', is_flag=True, help='Force remote temp table creation even table already exists in origin')
@click.option('--validate', is_flag=True, help='Validate the query sample')
@click.option('--force-copy-files', is_flag=True, help='Force the hdfs files copy procedure')
@click.option('--skip-remote-preparation', is_flag=True, help='Skip the creation of remote temp table')
@click.option('--engine', default=(os.path.relpath(".", "..")), help='Marvin engine name (default is the current folder)')
@click.option('--sql-id', '-q', help='If informed the process will be applied exclusivelly for this sample sql')
@click.option('--conf', '-c', default='hive_dataimport.conf', help='Hive data import configuration file')
@click.pass_context
def hive_dataimport_cli(
    ctx, conf, sql_id, engine, skip_remote_preparation, force_copy_files, validate, force,
    force_remote, max_query_size, destination_host, destination_port, destination_host_username,
    destination_host_password, destination_hdfs_root_path
):
    hive_dataimport(
        ctx, conf, sql_id, engine, skip_remote_preparation, force_copy_files, validate, force,
        force_remote, max_query_size, destination_host, destination_port, destination_host_username,
        destination_host_password, destination_hdfs_root_path
    )

def hive_dataimport(
    ctx, conf, sql_id, engine, skip_remote_preparation, force_copy_files, validate, force,
    force_remote, max_query_size, destination_host, destination_port, destination_host_username,
    destination_host_password, destination_hdfs_root_path
):

    initial_start_time = time.time()

    confs = read_config(filename=conf)

    if confs:
        print(chr(27) + "[2J")

        if sql_id:
            confs = [x for x in confs if x['sql_id'] == sql_id]

        for conf in confs:
            hdi = HiveDataImporter(
                max_query_size=max_query_size,
                destination_host=destination_host,
                destination_port=destination_port,
                destination_host_username=destination_host_username,
                destination_host_password=destination_host_password,
                destination_hdfs_root_path=destination_hdfs_root_path,
                engine=engine,
                **conf)

            if force:
                table_exists = False

            else:
                table_exists = hdi.table_exists(host=hdi.destination_host, db=hdi.origin_db, table=hdi.target_table_name)

            if not table_exists:
                hdi.import_sample(
                    create_temp_table=(not skip_remote_preparation),
                    copy_files=force_copy_files,
                    validate_query=validate,
                    force_create_remote_table=force_remote,
                )

            else:
                print ("Table {} already exists, skiping data import. Use --force flag to force data importation".format(hdi.full_table_name))

        print("Total Time : {:.2f}s".format(time.time() - initial_start_time))

        print("\n")


def read_config(filename):
    fname = os.path.join("", filename)
    if os.path.exists(fname):
        with open(fname, 'r') as fp:
            return json.load(fp)
    else:
        print("Configuration file {} doesn't exists...".format(filename))
        return {}


class HiveDataImporter():
    def __init__(
        self, origin_host, origin_db, origin_queue, target_table_name, sample_sql, engine,
        max_query_size, destination_host, destination_port, destination_host_username, destination_host_password,
        destination_hdfs_root_path, sql_id
    ):

        self.sql_id = sql_id
        self.origin_host = origin_host
        self.origin_db = origin_db
        self.origin_queue = origin_queue
        self.target_table_name = target_table_name
        self.sample_sql = sample_sql
        self.engine = engine
        self.destination_host = destination_host
        self.destination_port = destination_port
        self.destination_host_username = destination_host_username
        self.destination_host_password = destination_host_password
        self.destination_hdfs_root_path = destination_hdfs_root_path

        self.temp_db_name = 'marvin'
        self.max_query_size = max_query_size

        self.supported_format_types = {
            'TextInputFormat': 'TEXTFILE',
            'SequenceFileInputFormat': 'SEQUENCEFILE',
            'OrcInputFormat': 'ORC',
            'MapredParquetInputFormat': 'PARQUET',
            'AvroContainerInputFormat': 'AVRO',
            'RCFileInputFormat': 'RCFILE'
        }

        print("\n------------------------------------------------------------------------------")
        print("Initializing process for sql_id [{}]:".format(self.sql_id))
        print("     Origin -->")
        print("         Host:       [{}]".format(self.origin_host))
        print("         DataBase:   [{}]".format(self.origin_db))
        print("         Table Name: [{}]".format(self.target_table_name))
        print("         Sample SQL: [{}]".format(self.sample_sql))
        print("\n")
        print("     Destination -->")
        print("         Host:       [{}]".format(self.destination_host))
        print("         DataBase:   [{}]".format(self.origin_db))
        print("         Table Name: [{}]".format(self.target_table_name))
        print("\n")

    def validade_query(self):
        # creating connections
        print("Connecting with {} database on {} .. ".format(self.origin_db, self.origin_host))
        conn_origin = self.get_connection(host=self.origin_host, db=self.origin_db, queue=self.origin_queue)

        print("Counting sample sql ...")
        total_rows = self.count_rows(conn=conn_origin, sql=self.sample_sql)
        print("Found [{}] rows!".format(total_rows))

        print("Retrieve data sample for query estimation reasons...")
        data_sample = self.retrieve_data_sample(conn=conn_origin, full_table_name=self.full_table_name)
        print("Calculated [{}] bytes per row!".format(data_sample['estimate_query_mean_per_line']))

        estimated_size = data_sample['estimate_query_mean_per_line'] * total_rows

        print ("Estimated query size is : {} bytes".format(estimated_size))
        print ("Max permited query size is: {} bytes".format(self.max_query_size))

        return estimated_size <= self.max_query_size

    def table_exists(self, host, db, table):
        print("Verifiying if table {}.{} exists on {} ...".format(db, table, host))
        local_conn = self.get_connection(host=host)
        cursor = local_conn.cursor()

        cursor.execute("SHOW DATABASES LIKE '{}'".format(db))
        dbs = cursor.fetchall()
        self.show_log(cursor)

        if not len(dbs) == 1:
            table_exists = False
        else:
            cursor.execute("USE {} ".format(db))

            cursor.execute("SHOW TABLES LIKE '{}'".format(table))
            tbs = cursor.fetchall()
            self.show_log(cursor)

            if not len(tbs) == 1:
                table_exists = False
            else:
                table_exists = True

        cursor.close()
        return table_exists

    def reset_remote_tables(self):
        self.print_start_step(name="Reset Remote Tables for {}".format(self.temp_table_prefix), step_number=1, total_steps=1)

        print("Connecting with {} database on {} .. ".format(self.temp_db_name, self.origin_host))
        remote_temp_db_conn = self.get_connection(host=self.origin_host, db=self.temp_db_name, queue=self.origin_queue)

        cursor = remote_temp_db_conn.cursor()
        cursor.execute("SHOW TABLES LIKE '{}*'".format(self.temp_table_prefix))
        tbs = cursor.fetchall()
        self.show_log(cursor)
        cursor.close()

        valid_tbs = [tb[0] for tb in tbs]

        if valid_tbs:
            print("Found {} tables for deletion....".format(len(tbs)))

            for tb in valid_tbs:
                table_name = "{}.{}".format(self.temp_db_name, tb)
                print("Dropping table {} on {} .. ".format(table_name, self.origin_host))
                self.drop_table(conn=remote_temp_db_conn, table_name=table_name)

                hdfs_location = self.generate_table_location(self.destination_hdfs_root_path, self.origin_host, self.temp_db_name + '.db', tb)
                print("Removing hdfs files from {} .. ".format(hdfs_location))

                ssh = self._get_ssh_client(self.origin_host, self.destination_host_username, self.destination_host_password)
                self.delete_files(ssh, hdfs_location)

        else:
            print("No table found! Skiping reset remote tables process!!")

        self.print_finish_step()

    def print_finish_step(self):
        print("\n                                               STEP TAKES {:.4f} (seconds) ".format((time.time() - self.start_time)))

    def print_start_step(self, name, step_number, total_steps):
        print("\n------------------------------------------------------------------------------")
        print("MARVIN DATA IMPORT - STEP ({}) of ({}) - [{}]".format(step_number, total_steps, name))
        print("------------------------------------------------------------------------------\n")
        self.start_time = time.time()

    def import_sample(self, create_temp_table=True, copy_files=True, validate_query=True, force_create_remote_table=False):
        #
        #################################################################################
        # Step 1 - Query validation
        self.print_start_step(name="Query Validation", step_number=1, total_steps=6)

        is_valid = self.validade_query() if validate_query else True

        if not is_valid:
            print("Informed sample query is not valid!")
            self.print_finish_step()
            return

        self.print_finish_step()

        #
        ##################################################################################
        # Step 2 - Testing remote connecitons and getting table schema
        self.print_start_step(name="Table Schema Achievement", step_number=2, total_steps=6)

        # creating connections
        print("Connecting with {} database on {} .. ".format(self.origin_db, self.origin_host))
        conn_origin = self.get_connection(host=self.origin_host, db=self.origin_db, queue=self.origin_queue)

        print("Connecting with {} database on {} .. ".format(self.temp_db_name, self.origin_host))
        remote_temp_db_conn = self.get_connection(host=self.origin_host, db=self.temp_db_name, queue=self.origin_queue)

        # getting ddl from real table
        print("Getting DDL from {} table ".format(self.target_table_name))
        ddl = self.get_createtable_ddl(conn=conn_origin, origin_table_name=self.target_table_name, dest_table_name=self.temp_table_name)

        # validanting if partitions is used in query statement
        partitions = self.get_partitions(ddl)

        if validate_query and self.has_partitions(self.sample_sql, [p['cols'] for p in partitions]):
            print("Informed sample query doesn't have valid partitions in the clausule where!!!! Informe at lest one partition.")
            print("To disable the query validation use --skip-validation flag.")
            self.print_finish_step()
            return

        print("Connecting with DEFAULT database on {} .. ".format(self.destination_host))
        local_conn = self.get_connection(host=self.destination_host)

        # creating databases if not exists
        print("Creating database {} ...".format(self.origin_db))
        self.create_database(conn=local_conn, db=self.origin_db)

        print("Connecting with {} database on {} .. ".format(self.origin_db, self.destination_host))
        local_conn = self.get_connection(host=self.destination_host, db=self.origin_db)

        # creating databases if not exists
        print("Creating database {} ...".format(self.temp_db_name))
        self.create_database(conn=local_conn, db=self.temp_db_name)

        print("Connecting with {} database on {} .. ".format(self.temp_db_name, self.destination_host))
        local_temp_db_conn = self.get_connection(host=self.destination_host, db=self.temp_db_name)

        self.print_finish_step()

        #
        ##################################################################################
        # Step 3 - Remote Table Preparation
        self.print_start_step(name="Remote Table Preparation", step_number=3, total_steps=6)

        if create_temp_table:

            if force_create_remote_table:
                remote_table_exists = False

            else:
                remote_table_exists = self.table_exists(host=self.origin_host, db=self.temp_db_name, table=self.temp_table_name)

            # verify if remote table alredy exists
            if not remote_table_exists:
                print("Dropping table {} on {} .. ".format(self.full_temp_table_name, self.origin_host))
                self.drop_table(conn=remote_temp_db_conn, table_name=self.full_temp_table_name)

                print("Creating table {} on {} .. ".format(self.full_temp_table_name, self.origin_host))
                self.create_table(conn=remote_temp_db_conn, table_name=self.full_temp_table_name, ddl=ddl)

                # insert from select
                print("Populating table {} on {} using informed sample sql.. ".format(self.full_temp_table_name, self.origin_host))
                self.populate_table(conn=conn_origin, table_name=self.full_temp_table_name, partitions=partitions, sql=self.sample_sql)

            else:
                print("Table {} on {} already exists ...".format(self.full_temp_table_name, self.origin_host))

        self.print_finish_step()

        #
        ##################################################################################
        # Step 4 - Copying remote hdfs files
        self.print_start_step(name="Copying HDFS Files", step_number=4, total_steps=6)

        # get temp location
        print("Getting hdfs files location from {} table ...".format(self.full_temp_table_name))
        temp_table_location = self.get_table_location(conn=remote_temp_db_conn, table_name=self.full_temp_table_name)

        # copy hdfs files for local hdfs
        external_table_location = self.generate_table_location(
            host=self.destination_host,
            root_path=self.destination_hdfs_root_path,
            db_name=self.temp_db_name, table_name=self.temp_table_name)

        print("Copying files from [{}] to [{}]".format(temp_table_location, external_table_location))
        self.hdfs_dist_copy(force=copy_files,
                            hdfs_host=self.destination_host,
                            hdfs_port=self.destination_port,
                            origin=temp_table_location,
                            dest=external_table_location,
                            password=self.destination_host_password,
                            username=self.destination_host_username)

        self.print_finish_step()
        #
        ##################################################################################
        # Step 5 - External table creation using hdfs files
        self.print_start_step(name="Local Temporary Table Creation", step_number=5, total_steps=6)

        # creating external table using parquet files in hdfs
        print("Dropping temp table {} on {} .. ".format(self.full_temp_table_name, self.destination_host))
        self.drop_table(conn=local_temp_db_conn, table_name=self.full_temp_table_name)

        # create temp table
        print("Creating temp table {} using imported hdfs files from [{}] ...".format(self.full_temp_table_name, external_table_location))
        self.create_external_table(conn=local_temp_db_conn,
                                   temp_table_name=self.full_temp_table_name,
                                   ddl=ddl,
                                   parquet_file_location=external_table_location)

        print("Refreshing table {} partitions on {} ..".format(self.full_temp_table_name, self.destination_host))
        self.refresh_partitions(conn=local_temp_db_conn, table_name=self.full_temp_table_name)

        self.print_finish_step()

        #
        ##################################################################################
        # Step 6 - Destination table creation from external table
        self.print_start_step(name="Table population", step_number=6, total_steps=6)

        # create view
        print("Dropping table view {} on {} .. ".format(self.full_table_name, self.destination_host))
        self.drop_view(conn=local_conn, view_name=self.full_table_name)

        print("Creating table view {} ... ".format(self.full_table_name, self.destination_host))
        self.create_view(conn=local_conn, view_name=self.full_table_name, table_name=self.full_temp_table_name)

        self.print_finish_step()

        print("Procedure done!!!!")

    @property
    def temp_table_prefix(self):
        return "{}".format(slugify(self.engine).replace('-', '_'))

    @property
    def temp_table_name(self):
        return "{}_{}_{}_{}".format(
            self.temp_table_prefix,
            self.origin_db,
            self.target_table_name,
            hashlib.sha1(slugify(self.sample_sql).encode('utf-8')).hexdigest()
        )

    @property
    def full_table_name(self):
        return "{}.{}".format(self.origin_db, self.target_table_name)

    @property
    def full_temp_table_name(self):
        return "{}.{}".format(self.temp_db_name, self.temp_table_name)

    def generate_table_location(self, root_path, host, db_name, table_name):
        return "hdfs://{}:8020{}".format(host, os.path.join(root_path, db_name, table_name))

    def clean_ddl(self, ddl, remove_formats=True, remove_general=True):
        if remove_general:
            # Removing LOCATION statement
            regex = "(LOCATION\s+'(.*?)')"
            result = re.search(regex, ddl)
            ddl = ddl.replace(result.group(1), " ") if result else ddl

            # Removing TBLPROPERTIES statement
            regex = "(TBLPROPERTIES\s+(.*?)\))"
            result = re.search(regex, ddl)
            ddl = ddl.replace(result.group(1), " ") if result else ddl

            # Removing WITH SERDEPROPERTIES statement
            regex = "(WITH SERDEPROPERTIES\s+(.*?)\))"
            result = re.search(regex, ddl)
            ddl = ddl.replace(result.group(1), " ") if result else ddl

        if remove_formats:
            # Removing STORED AS INPUTFORMAT statement
            regex = "(STORED AS INPUTFORMAT\s+'(.*?)')"
            result = re.search(regex, ddl)
            ddl = ddl.replace(result.group(1), " ") if result else ddl

            # Removing OUTPUTFORMAT statement
            regex = "(OUTPUTFORMAT\s+'(.*?)')"
            result = re.search(regex, ddl)
            ddl = ddl.replace(result.group(1), " ") if result else ddl

        return ddl

    def get_table_format(self, ddl):
        regex = "(STORED AS INPUTFORMAT\s+'(.*?)')"
        result = re.search(regex, ddl)
        input_format = result.group(2)
        return self.supported_format_types[input_format.split(".")[-1]]

    def get_database_info(self, ddl):
        regex = "CREATE TABLE `((.*?)\.)?(.*?)`\("
        result = re.search(regex, ddl)
        if result:
            groups = result.groups()
            if groups[0]:
                # found db name
                return {'db': groups[1], 'table': groups[2]}
            else:
                {'db': None, 'table': groups[2]}
        return {'db': None, 'table': None}

    def get_createtable_ddl(self, conn, origin_table_name, dest_table_name):
        cursor = conn.cursor()
        cursor.execute("SHOW CREATE TABLE " + origin_table_name)
        _lines = [_line[0] for _line in cursor.fetchall()]
        ddl = ''.join(_lines)
        ddl = self.clean_ddl(ddl, remove_formats=False, remove_general=True)
        ddl = ddl.replace(origin_table_name, dest_table_name)
        cursor.close()
        return ddl

    def create_database(self, conn, db):
        self._execute_db_command(conn, "CREATE DATABASE IF NOT EXISTS " + db)

    def drop_table(self, conn, table_name):
        self._execute_db_command(conn, 'DROP TABLE IF EXISTS ' + table_name)

    def drop_view(self, conn, view_name):
        self._execute_db_command(conn, 'DROP VIEW ' + view_name)

    def create_table(self, conn, table_name, ddl):
        self._execute_db_command(conn, ddl)

    def _execute_db_command(self, conn, command):
        cursor = conn.cursor()
        cursor.execute(command)
        self.show_log(cursor)
        cursor.close()

    def get_connection(self, host, db='DEFAULT', queue='default'):
        return hive.connect(host=host,
                            database=db,
                            configuration={'mapred.job.queue.name': queue,
                                           ' hive.exec.dynamic.partition.mode': 'nonstrict'})

    def retrieve_data_sample(self, conn, full_table_name, sample_limit=100):
        cursor = conn.cursor()

        sql = "SELECT * FROM {} TABLESAMPLE ({} ROWS)".format(full_table_name, sample_limit)

        cursor.execute(sql)
        data_header = [{'col': line[0].split('.')[1], 'table': line[0].split('.')[0], 'type': line[1]} for line in cursor.description]
        data = [row for row in cursor.fetchall()]
        self.show_log(cursor)
        cursor.close()
        return {'data_header': data_header,
                'total_lines': len(data),
                'data': data,
                'estimate_query_size': sys.getsizeof(data),
                'estimate_query_mean_per_line': sys.getsizeof(data) / len(data)}

    def count_rows(self, conn, sql):
        cursor = conn.cursor()
        cursor.execute("SELECT COUNT(1) " + sql[sql.upper().rfind("FROM"):])
        size = cursor.fetchone()[0]
        self.show_log(cursor)
        cursor.close()
        return size

    def show_log(self, cursor):
        for l in cursor.fetch_logs():
            logger.debug(l)

    def save_data(self, conn, table, data):
        cursor = conn.cursor()
        print('Inserting {} rows in {} table...'.format(data['total_lines'], table))
        cols = [v['col'] for v in data['data_header']]
        dml = "INSERT INTO {0} ({1}) VALUES ({2})".format(table, ", ".join(cols), ", ".join(['%s' for col in cols]))
        cursor.executemany(dml, [(v,) for v in data['data'][1:10]])
        self.show_log(cursor)
        cursor.close()

    def get_partitions(self, ddl):
        regex = "(PARTITIONED BY\s+\((.*?)\))"
        result = re.search(regex, ddl)
        if result:
            p_cols = result.group(2).strip().replace('`', '').split(",")
            return [{'col': p_col.split()[0], 'type': p_col.split()[1]} for p_col in p_cols]
        else:
            return []

    def has_partitions(self, sql, partitions):
        regex = "WHERE(.*?)(" + "|".join(partitions).upper() + ")"
        result = re.search(regex, sql.upper())

        if result:
            return True
        else:
            return False

    def populate_table(self, conn, table_name, partitions, sql):
        partitions = [p['col'] for p in partitions]
        partitions_statement = "PARTITION ({})".format(", ".join(partitions)) if partitions else ""
        dml = "INSERT OVERWRITE TABLE {0} {1} {2}".format(table_name, partitions_statement, sql)
        self._execute_db_command(conn, dml)

    def create_view(self, conn, view_name, table_name):
        dml = "CREATE VIEW {0} AS SELECT * FROM {1}".format(view_name, table_name)
        self._execute_db_command(conn, dml)

    def refresh_partitions(self, conn, table_name):
        refresh_statement = "MSCK REPAIR TABLE {0}".format(table_name)
        self._execute_db_command(conn, refresh_statement)

    def get_table_location(self, conn, table_name):
        cursor = conn.cursor()
        cursor.execute("DESCRIBE FORMATTED {}".format(table_name))
        location = [key[1].strip() for key in cursor.fetchall() if key[0] and key[0].strip().upper() == 'LOCATION:']
        location = location[0].replace('hdfs://', 'hftp://')
        cursor.close()
        return location

    def delete_files(self, ssh, url):
        cmd = "hdfs dfs -rm -R '{}'".format(url)
        self._hdfs_commands(ssh, cmd)

    def copy_files(self, ssh, origin, dest):
        cmd = "hadoop distcp --update '{}' '{}'".format(origin, dest)
        return self._hdfs_commands(ssh, cmd)

    def _hdfs_commands(self, ssh, cmd):
        logger.debug("Executing remote command: {}".format(cmd))
        i, o, e = ssh.exec_command(cmd)
        errors = e.readlines()
        output = o.readlines()
        logger.debug(output)
        logger.debug(errors)
        return output, errors

    def _get_ssh_client(self, hdfs_host, hdfs_port, username, password):
        ssh = SSHClient()
        ssh.set_missing_host_key_policy(AutoAddPolicy())
        ssh.connect(hostname=hdfs_host, port=hdfs_port, username=username, password=password, )
        return ssh

    def hdfs_dist_copy(self, force, hdfs_host, hdfs_port, origin, dest, username=None, password=None):
        # connecting with hdfs host
        ssh = self._get_ssh_client(hdfs_host, hdfs_port, username, password)

        if force:
            print("Removing old hdfs files if necessary. To force copy remote files use --force-copy-files flag.")

            # delete files from dest
            self.delete_files(ssh, dest)

        else:
            print("Using old hdfs files to complete the procedure. If necessary to copy files again use --force-copy-files flag.")

        # copy files from origin to destination
        _, copy_errors = self.copy_files(ssh, origin, dest)

        # validate copy
        cmd_template = "hdfs dfs -ls -R '{}' | grep -E '^-' | wc -l"
        cmd = cmd_template.format(origin)
        result1, _ = self._hdfs_commands(ssh, cmd)

        cmd = cmd_template.format(dest)
        result2, _ = self._hdfs_commands(ssh, cmd)

        if result1 == result2:
            print("Files {} successfully transferred!!".format(result1))
        else:
            print("Errors during hdfs files copy process!!")
            for e_l in copy_errors:
                logger.debug(e_l)
            sys.exit("Stoping process!")

    def create_external_table(self, conn, temp_table_name, ddl, parquet_file_location):
        format_type = self.get_table_format(ddl)
        ddl = self.clean_ddl(ddl, remove_formats=True, remove_general=False)
        ddl = ddl.replace("CREATE TABLE", "CREATE EXTERNAL TABLE")
        ddl = "{} STORED AS {} LOCATION '{}'".format(ddl, format_type, parquet_file_location)
        self.create_table(conn=conn, table_name=temp_table_name, ddl=ddl)
