airflow/hooks/hdfs_hook.py
def get_conn(self):
        """
        Returns a snakebite HDFSClient object.
        """
        # When using HAClient, proxy_user must be the same, so is ok to always
        # take the first.
        effective_user = self.proxy_user
        autoconfig = self.autoconfig
        use_sasl = configuration.conf.get('core', 'security') == 'kerberos'

        try:
            connections = self.get_connections(self.hdfs_conn_id)

            if not effective_user:
                effective_user = connections[0].login
            if not autoconfig:
                autoconfig = connections[0].extra_dejson.get('autoconfig',
                                                             False)
            hdfs_namenode_principal = connections[0].extra_dejson.get(
                'hdfs_namenode_principal')
        except AirflowException:
            if not autoconfig:
                raise

        if autoconfig:
            # will read config info from $HADOOP_HOME conf files
            client = AutoConfigClient(effective_user=effective_user,
                                      use_sasl=use_sasl)
        elif len(connections) == 1:
            client = Client(connections[0].host, connections[0].port,
                            effective_user=effective_user, use_sasl=use_sasl,
                            hdfs_namenode_principal=hdfs_namenode_principal)
        elif len(connections) > 1:
            nn = [Namenode(conn.host, conn.port) for conn in connections]
            client = HAClient(nn, effective_user=effective_user,
                              use_sasl=use_sasl,
                              hdfs_namenode_principal=hdfs_namenode_principal)
        else:
            raise HDFSHookException("conn_id doesn't exist in the repository "
                                    "and autoconfig is not specified")

        return client
def get_conn(self):
        # When using HAClient, proxy_user must be the same, so is ok to always
        # take the first.
        effective_user = self.proxy_user
        autoconfig = self.autoconfig
        use_sasl = configuration.conf.get('core', 'security') == 'kerberos'

        try:
            connections = self.get_connections(self.hdfs_conn_id)

            if not effective_user:
                effective_user = connections[0].login
            if not autoconfig:
                autoconfig = connections[0].extra_dejson.get('autoconfig',
                                                             False)
            hdfs_namenode_principal = connections[0].extra_dejson.get(
                'hdfs_namenode_principal')
        except AirflowException:
            if not autoconfig:
                raise

        if autoconfig:
            # will read config info from $HADOOP_HOME conf files
            client = AutoConfigClient(effective_user=effective_user,
                                      use_sasl=use_sasl)
        elif len(connections) == 1:
            client = Client(connections[0].host, connections[0].port,
                            effective_user=effective_user, use_sasl=use_sasl,
                            hdfs_namenode_principal=hdfs_namenode_principal)
        elif len(connections) > 1:
            nn = [Namenode(conn.host, conn.port) for conn in connections]
            client = HAClient(nn, effective_user=effective_user,
                              use_sasl=use_sasl,
                              hdfs_namenode_principal=hdfs_namenode_principal)
        else:
            raise HDFSHookException("conn_id doesn't exist in the repository "
                                    "and autoconfig is not specified")

        return client
def get_conn ( self ) : # When using HAClient, proxy_user must be the same, so is ok to always # take the first. effective_user = self . proxy_user autoconfig = self . autoconfig use_sasl = configuration . conf . get ( 'core' , 'security' ) == 'kerberos' try : connections = self . get_connections ( self . hdfs_conn_id ) if not effective_user : effective_user = connections [ 0 ] . login if not autoconfig : autoconfig = connections [ 0 ] . extra_dejson . get ( 'autoconfig' , False ) hdfs_namenode_principal = connections [ 0 ] . extra_dejson . get ( 'hdfs_namenode_principal' ) except AirflowException : if not autoconfig : raise if autoconfig : # will read config info from $HADOOP_HOME conf files client = AutoConfigClient ( effective_user = effective_user , use_sasl = use_sasl ) elif len ( connections ) == 1 : client = Client ( connections [ 0 ] . host , connections [ 0 ] . port , effective_user = effective_user , use_sasl = use_sasl , hdfs_namenode_principal = hdfs_namenode_principal ) elif len ( connections ) > 1 : nn = [ Namenode ( conn . host , conn . port ) for conn in connections ] client = HAClient ( nn , effective_user = effective_user , use_sasl = use_sasl , hdfs_namenode_principal = hdfs_namenode_principal ) else : raise HDFSHookException ( "conn_id doesn't exist in the repository " "and autoconfig is not specified" ) return client
Returns a snakebite HDFSClient object.
Returns a snakebite HDFSClient object .