HDFS / Impala with high availability

Preamble

This article applies when using Python on HDFS with an high availability option. The special characteristic of high availability is to have two namenodes for a single HDFS, used in case of failure.

At then end of the article you'll find an exemple on how to connect to a random active Data Node for Impala. It's helpful to split the work load on all data nodes and to connect to an active data node to avoid the failure of your job.

Libraries dependency

from hdfs import InsecureClient

Function returning the active NameNode

def get_url_active_namenode(user, list_name_nodes, port_hdfs="50070"):
    cpt = 1
    # Loop to identify which one of the 2 NameNode is active
    for url in list_name_nodes:
        # Create the HDFS client
        hadoop = InsecureClient(url + ':' + str(port_hdfs), user=user)
        try:
            # Test if the HDFS client is working
            hadoop.status('/')
            return url
        except:
            # if an error occure, it means the NameNode used to connect to HDFS is not active
            if cpt == len(HDFS_URLS):
                # if we tested both NameNodes, it means there is no active NameNode
                # HDFS is not reachable
                raise NameError("No NameNode available")
            else:
                cpt += 1

Example:

list_name_nodes = ["<url_of_namenode_1>", "<url_of_namenode_2>"]
user= "user_test"
url = get_url_active_namenode(user, list_name_nodes)

Function returning an HDFS client

def return_client_hdfs(user, list_name_nodes, port_hdfs="50070"):
    url = get_url_active_namenode(user, list_name_nodes, port_hdfs)
    client_hdfs = InsecureClient(url, user=user)
    return client_hdfs

Example:

list_name_nodes = ["<url_of_namenode_1>", "<url_of_namenode_2>"]
user= "user_test"
client_hdfs = return_client_hdfs(user=user, list_name_nodes=list_name_nodes)

Function returning an active random DataNode for Impala

# os.environ['PORT_HDFS'] = 50070
# os.environ['PORT_IMPALA'] = 21050

def get_client_ibis_impala(user, user_password, list_name_nodes, list_data_nodes):
    # Get The active NameNode
    url_hdfs = get_url_active_namenode(user, list_name_nodes, port_hdfs=os.environ['PORT_HDFS'])

    # Use the URL of the active NameNode to create an IBIS HDFS client
    ibis_hdfs = ibis.impala.hdfs_connect(host=url_hdfs, port=int(os.environ['PORT_HDFS']))

    # Loop through all Data Nodes to test if they are active or not
    while len(list_data_nodes) > 0:
        # Randomly select one the DataNode
        data_node = list_data_nodes[random.randint(0, len(list_data_nodes) - 1)]
        try:
            # Test if the DataNode is active
            ibis_client = ibis.impala.connect(host=data_node, port=int(os.environ['PORT_IMPALA']),
                                              hdfs_client=ibis_hdfs,
                                              user=user,
                                              password=user_password,
                                              auth_mechanism='PLAIN',
                                              timeout=0.5)
            ibis_client.list_databases()
            return ibis_client

        except:
            # If an error occure, it means the DataNode is not ative or does not exists
            # -> We need to remove that DataNode from our list to avoir infinite loop and testing a DataNode twice
            data_node_list.remove(data_node)

    raise NameError("No DataNode available for Impala")

Example:

user="user_test"
pwd = "<pwd_of_user_test>"
list_name_nodes = ["<url_of_namenode_1>", "<url_of_namenode_2>"]
list_data_nodes = ["<url_of_datanode_1>", "<url_of_datanode_2>", "<url_of_datanode_3>"]
client_impala = get_client_ibis_impala(user , pwd, list_name_nodes, list_data_nodes)