File Coverage

blib/lib/Apache/Hadoop/Config.pm

Criterion	Covered	Total	%
statement	8	164	4.8
branch	0	62	0.0
condition	0	39	0.0
subroutine	3	25	12.0
pod	0	9	0.0
total	11	299	3.6

package Apache::Hadoop::Config;

use 5.010001;
use strict;
use warnings;

our @ISA = qw();
our $VERSION = '0.01';


# class definition begins

sub new {
    my $class = shift;
    my %args  = @_;
    my $self  = {
        namenode       => $args{'namenode'}  || 'localhost',
        secondary      => $args{'secondary'} || '0.0.0.0',
        proxynode      => $args{'proxynode'} || 'localhost',
        proxyport      => $args{'proxyport'} || 8888,
        hadoop_install => $args{'hadoop_install'}  || '/usr/local/hadoop',
        hadoop_confdir => $args{'hadoop_confdir'}  || '/usr/local/hadoop/etc/hadoop',
        hdfs_tmp       => $args{'hdfs_tmp'}  || '/hdfs/tmp',
        hdfs_name_disks=> $args{'hdfs_name_disks'} ||
            [ '/hdfs/name1',
              '/hdfs/name2',
            ],
        hdfs_data_disks=> $args{'hdfs_data_disks'} ||
            [ '/hdfs/data1',
              '/hdfs/data2',
              '/hdfs/data3',
              '/hdfs/data4',
            ],
        hadoop_logs    => [ '/logs', '/logs/userlog' ],
        debug          => $args{'debug'}    || undef,

        config         => $args{'config'}   || undef,

        sysinfo=> {
            cpu        => $args{'cpuinfo'}  || undef,
            mem        => $args{'meminfo'}  || undef,
            disk       => $args{'diskinfo'} || undef,
        },
    };
    bless $self, $class;
    return $self;
}

# internal utils
sub _minimum {
    my ($self, @arr) = (@_);
    my $min;
    for (@arr) { $min = $_ if !$min || $_ < $min; }
    return $min;
}

sub _maximum {
    my ($self, @arr) = (@_);
    my $max;
    for (@arr) { $max = $_ if !$max || $_ > $max; }
    return $max;
}

sub _copyconf {
    my ($self, $config) = (@_);
    unless ( defined $self->{'config'} ) {
        $self->{'config'} = $config;
        return;
    }

    foreach my $file ( keys %{$config} ) {
        foreach my $param ( keys %{$config->{$file}} ) {
            $self->{'config'}->{$file}->{$param} = $config->{$file}->{$param}
                unless defined $self->{'config'}->{$file}->{$param};
        }
    }
}

# get config template
sub basic_config {
    my ($self) = (@_);
    my $config = {
        'core-site.xml' => {
            'fs.defaultFS' => 'http://'.$self->{namenode}.':9000',
            'hadoop.tmp.dir' => $self->{'hdfs_tmp'},
            },
        'hdfs-site.xml' => {
            'dfs.replication' => 1,
            'dfs.namenode.name.dir' => join ( ',', map { 'file://'.$_ } @{$self->{'hdfs_name_disks'}} ),
            'dfs.datanode.data.dir' => join ( ',', map { 'file://'.$_ } @{$self->{'hdfs_data_disks'}} ),

            # secondary namenode
            'dfs.namenode.secondary.http-address' => $self->{'secondary'}.':50090',
            'dfs.namenode.secondary.https-address'=> $self->{'secondary'}.':50091',
            },
        'yarn-site.xml' => {
            'yarn.nodemanager.aux-services' => 'mapreduce_shuffle',
            'yarn.nodemanager.aux-services.mapreduce.shuffle.class' => 'org.apache.hadoop.mapred.ShuffleHandler',
            'yarn.web-proxy.address' => $self->{'proxynode'}.':'.$self->{'proxyport'},
            },
        'mapred-site.xml' => {
            'mapreduce.framework.name' => 'yarn',
            }
        };

    $self->_copyconf ( $config );
}

#
# directory management
#
sub _mkdir {
    my ($self, %opts) = (@_);
    my $mode = $opts{'mode'} || 0750;
    my $u = umask (0);
    map { mkdir $_, $mode; chdir $_; } split (/\//, $opts{'directory'});
}

sub create_hdfs_name_disks {
    my ($self) = (@_);
    foreach my $dir ( @{$self->{'hdfs_name_disks'}} ) {
        print "creating ".$dir, "\n" if defined $self->{'debug'};
        $self->_mkdir( directory => $dir );
    }
}

sub create_hdfs_data_disks {
    my ($self) = (@_);
    foreach my $dir ( @{$self->{'hdfs_data_disks'}} ) {
        print "creating ".$dir, "\n" if defined $self->{'debug'};
        $self->_mkdir( directory => $dir );
    }
}

sub create_hdfs_tmpdir {
    my ($self) = (@_);
    print "creating ".$self->{'hdfs_tmp'}, "\n" if defined $self->{'debug'};
    $self->_mkdir( directory => $self->{'hdfs_tmp'}, mode => 01775 );
}

sub create_hadoop_logdir {
    # $hadoop_install/tmp 1775
    my ($self) = (@_);
    foreach my $dir ( @{$self->{'hadoop_logs'}} ) {
        print "creating ".$self->{'hadoop_install'}.$dir, "\n" if defined $self->{'debug'};
        $self->_mkdir( directory => $self->{'hadoop_install'}.$dir, mode => 01775 );
    }
}
# ends

#
# begin recommended settings
#

# get cpu core count
sub _get_cpu_cores {
    my ($self) = (@_);
    my $cpuinfo = '/proc/cpuinfo';
    open CPU, $cpuinfo or die "Cannot open $cpuinfo, $!\n";
    $self->{'sysinfo'}->{'cpu'} = scalar (map /^processor/, );
    close CPU;
}

# get memory size
sub _get_memory_system {
    my ($self) = (@_);
    my $meminfo = '/proc/meminfo';
    open MEM, $meminfo or die "Cannot open $meminfo, $!\n";
    $self->{'sysinfo'}->{'mem'} = (split (/\s+/, (grep /^MemTotal/, )[0]))[1];
    $self->{'sysinfo'}->{'mem'} = $self->{'sysinfo'}->{'mem'} / (1024.0 * 1024.0);
    close MEM;
}

# get disk info: not implemented
sub _get_disk_count {
    my ($self) = (@_);
    $self->{'sysinfo'}->{'disk'} = scalar @{$self->{'hdfs_data_disks'}}
}

# calculate reserved memory
sub _get_memory_reserved {
    my ($self) = (@_);
    my ($m, $m1, $m2, $m3);

    $self->_get_memory_system unless defined $self->{'sysinfo'}->{'mem'};
    $m = $self->{'sysinfo'}->{'mem'};
    $m1 = int (0.124102 * $m + 1.236659); # approximated by a linear eq through least square
    $m2 = $m >= 16 ? $m1 - ($m1%2) : $m1;
    $m3 = $m >= 64 ? $m2 - ($m2%2) : $m2;
    print "reserved_mem (gb): ",$m3,"\n" if $self->{'debug'};
    return $m3;
}

# calculate available memory (total - reserved)
sub _calc_memory_available {
    my ($self) = (@_);
    $self->_get_memory_system unless defined $self->{'sysinfo'}->{'mem'};
    my $mem = $self->{'sysinfo'}->{'mem'} - $self->_get_memory_reserved;
    $mem *= 1024; # mb
    print "available mem (mb): $mem\n" if $self->{'debug'};
    return $mem;
}

# calculate min container size
sub _calc_min_container_size {
    my ($self) = (@_);
    $self->_get_memory_system unless defined $self->{'sysinfo'}->{'mem'};
    my $m = $self->{'sysinfo'}->{'mem'};
    my $m1 = $m < 4 ? 0 : ($m < 8 ? 1 : ($m<24 ? 2 : 3));
    my $m2 = 2 ** $m1;
    my $m3 = 0.25 * $m2 * 1024; # mb
    print "min_container_size (mb): $m3\n" if $self->{'debug'};
    return $m3;
}

sub _calc_container_count {
    my ($self) = (@_);
    my ($cores, $disks);
    my $ratio = $self->_calc_memory_available / $self->_calc_min_container_size;

    $self->_get_cpu_cores unless defined $self->{'sysinfo'}->{'cpu'};
    $cores = 2 * $self->{'sysinfo'}->{'cpu'};

    unless (defined $self->{'sysinfo'}->{'disk'}) { $self->_get_disk_count; }
    $disks = 1.8 * $self->{'sysinfo'}->{'disk'};

    print "cores=$cores disks=$disks ratio=$ratio\n" if $self->{'debug'};
    return int ($self->_minimum($cores, $disks, $ratio));
}

sub _calc_container_memory {
    my ($self) = (@_);
    my $ratio = $self->_calc_memory_available / $self->_calc_container_count;
    my $size  = $self->_calc_min_container_size;
    return int($self->_maximum($size, $ratio));
}

sub memory_config {
    my ($self) = (@_);

    my $con = $self->_calc_container_count;
    my $mpc = $self->_calc_container_memory;

    my $config = {
        'yarn-site.xml' => {
            'yarn.nodemanager.resource.memory-mb' => $con * $mpc,
            'yarn.scheduler.minimum-allocation-mb' => $mpc,
            'yarn.scheduler.maximum-allocation-mb' => $con * $mpc,
            },
        'mapred-site.xml' => {
            'mapreduce.map.memory.mb' => $mpc,
            'mapreduce.reduce.memory.mb' => 2 * $mpc,
            'mapreduce.map.java.opts' => "-Xmx".int(0.8 * $mpc)."m",
            'mapreduce.reduce.java.opts' => "-Xmx".int(0.8 * 2 * $mpc)."m",
            },
        };
    $self->_copyconf ( $config );
}

sub print_config {
    my ($self) = (@_);

    print "min cont size (mb)    : ",$self->_calc_min_container_size,"\n";
    print "num of containers     : ",$self->_calc_container_count,"\n";
    print "mem per container (mb): ",$self->_calc_container_memory,"\n";
    foreach my $k ( keys %{$self->{'sysinfo'}} ) {
        print sprintf "%5s : %s\n", $k, $self->{'sysinfo'}->{$k}
            if defined $self->{'sysinfo'}->{$k};
    }

    print "---------------\n";
    foreach my $file ( keys %{$self->{'config'}} ) {
        print $file, "\n";
        my $val = $self->{'config'}->{$file};
        foreach my $key ( keys %$val ) {
            print "  ",$key,": ", $val->{$key},"\n";
        }
    }
    print "---------------\n";
}

sub _tag {
    my ($self, %args) = (@_);

    my $header = qq{




};

    my $property = qq{
    NAME
    VALUE

};
    # type = header, config, property
    # (type=config) begin=>1, end=>1
    # (type==property) name=>s value=>s
    #
    for ($args{'type'}) {
        /header/   && do {
                        return $header;
                        };
        /config/   && do {
                        my $str;
                        $str .= "\n" if defined $args{'begin'};
                        $str .= "\n" if defined $args{'end'};
                        return $str;
                        };
        /property/ && do {
                        my $str = $property;
                        $str =~ s/NAME/$args{'name'}/;
                        $str =~ s/VALUE/$args{'value'}/;
                        return $str;
                        };
    }
}

sub write_config {
    my ($self, %args) = (@_);
    my $cdir = $args{'confdir'} || $self->{'confdir'};

    foreach my $file ( keys %{$self->{'config'}} ) {
        print "-> writing to $cdir/$file ...\n";

        # open conf file
        open CONF, ">".$cdir.'/'.$file or die $!;

        # xml header
        print CONF $self->_tag (type=>'header');
        print CONF $self->_tag (type=>'config', begin=>1);

        # properties
        my $prop = $self->{'config'}->{$file};
        foreach my $key ( keys %{$prop} ) {
            print CONF $self->_tag( type=>'property', name=>$key, value=>$prop->{$key} );
        }

        # close config
        print CONF $self->_tag (type=>'config', end=>1);
        close CONF;
    }
}
#
# ends recommended settings

1;
__END__