MiniYARNCluster MiniDFSClusterKerberos

/**

成都创新互联专注于万州企业网站建设,响应式网站设计,商城系统网站开发。万州网站建设公司,为万州等地区提供建站服务。全流程按需定制网站，专业设计，全程项目跟踪，成都创新互联专业和态度为您提供的服务

1. download hadoop source code  ,then yum install cmake zlib-devel,  finally ,mvn clean package -Pdist,native -DskipTests -Dtar to create native lib

2. before run junit you shuold specify vm argument  -Djava.library.path=/usr/hadoop{version}/hadoop-dist/target/hadoop-{version}/lib/native

3. use mvn clean install -Pnative -Dcontainer-executor.conf.dir=/{project.path}/bin   -DskipTests  to create container-executor

   an example container-executor.cfg existe in {project.path}/bin

   After that it will create  container-executor exe file

   cp this file to {project.path}/bin  

   sudo chown root:yourusername {project.path}/bin/container-executor in this example yourusername is tjj

   sudo chmod 4550 {project.path}/bin/container-executor

 4. If you want to run testjob in YARN(conf.set("mapreduce.framework.name", "yarn")),you should modify META-INF/services/org.apache.hadoop.mapreduce.protocol.ClientProtocolProvider

   to #org.apache.hadoop.mapred.LocalClientProtocolProvider

      org.apache.hadoop.mapred.YarnClientProtocolProvider

   in hadoop-mapreduce-client-common-{version}.jar which exist in your classpath

 */

import static org.apache.hadoop.fs.CommonConfigurationKeys.IPC_CLIENT_CONNECT_MAX_RETRIES_ON_SASL_KEY;

import static org.apache.hadoop.hdfs.DFSConfigKeys.*;

import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_HTTPS_ADDRESS_KEY;

import static org.junit.Assert.*;

import java.io.*;

import java.util.ArrayList;

import java.util.List;

import java.util.Properties;

import org.apache.commons.io.FileUtils;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.*;

import org.apache.hadoop.hdfs.HdfsConfiguration;

import org.apache.hadoop.hdfs.MiniDFSCluster;

import org.apache.hadoop.hdfs.protocol.datatransfer.sasl.SaslDataTransferTestCase;

import org.apache.hadoop.http.HttpConfig;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import org.apache.hadoop.minikdc.MiniKdc;

import org.apache.hadoop.security.SecurityUtil;

import org.apache.hadoop.security.UserGroupInformation;

import org.apache.hadoop.security.ssl.KeyStoreTestUtil;

import org.apache.hadoop.yarn.conf.YarnConfiguration;

import org.apache.hadoop.yarn.server.MiniYARNCluster;

import org.junit.*;

public class TestClusterWithKerberos {

    private static File baseDir;

    private static String hdfsPrincipal;

    private static MiniKdc kdc;

    private static String keytab;

    private static String spnegoPrincipal;

    private MiniYARNCluster yarnCluster;

    private MiniDFSCluster cluster;

    @BeforeClass

    public static void initKdc() throws Exception {

        baseDir = new File(System.getProperty("test.build.dir", "target/test-dir"),

                SaslDataTransferTestCase.class.getSimpleName());

        FileUtil.fullyDelete(baseDir);

        assertTrue(baseDir.mkdirs());

        Properties kdcConf = MiniKdc.createConf();

        kdc = new MiniKdc(kdcConf, baseDir);

        kdc.start();

        UserGroupInformation ugi = UserGroupInformation.createRemoteUser("tjj");

        UserGroupInformation.setLoginUser(ugi);

        String userName = UserGroupInformation.getLoginUser().getShortUserName();

        File keytabFile = new File(baseDir, userName + ".keytab");

        keytab = keytabFile.getAbsolutePath();

        kdc.createPrincipal(keytabFile, userName + "/localhost", "HTTP/localhost");

        hdfsPrincipal = userName + "/localhost@" + kdc.getRealm();

        spnegoPrincipal = "HTTP/localhost@" + kdc.getRealm();

        System.out.println("keytab "+keytab+"hdfsPrincipal "+hdfsPrincipal);

    }

    @AfterClass

    public static void shutdownKdc() {

        if (kdc != null) {

            kdc.stop();

        }

        FileUtil.fullyDelete(baseDir);

    }

    private void startCluster(HdfsConfiguration conf) throws IOException {

        cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build();//

        cluster.waitActive();

        yarnCluster = new MiniYARNCluster("MiniClusterStartsWithCountJobTest", // testName

                1, // number of node managers

                1, // number of local log dirs per node manager

                1); // number of hdfs dirs per node manager

        yarnCluster.init(conf);

        yarnCluster.start();

        yarnCluster.getConfig().writeXml(new FileOutputStream(new File("conf.Xml")));

    }

    @Test

    public void testWithMiniCluster() throws Exception {

        HdfsConfiguration clusterConf = createSecureConfig("authentication,integrity,privacy");

        YarnConfiguration yarnConf =  createYarnSecureConfig();

        clusterConf.addResource(yarnConf);

        startCluster(clusterConf);

        Configuration conf = new Configuration();

       conf.addResource(FileUtils.openInputStream(new File("conf.Xml")));

        String IN_DIR = "testing/wordcount/input";

        String OUT_DIR = "testing/wordcount/output";

        String DATA_FILE = "sample.txt";

        FileSystem fs = FileSystem.get(conf);

        Path inDir = new Path(IN_DIR);

        Path outDir = new Path(OUT_DIR);

        fs.delete(inDir, true);

        fs.delete(outDir, true);

        // create the input data files

        List content = new ArrayList();

        content.add("She sells seashells at the seashore, and she sells nuts in the mountain.");

        writeHDFSContent(fs, inDir, DATA_FILE, content);

        // set up the job, submit the job and wait for it complete

        Job job = Job.getInstance(conf);

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(IntWritable.class);

        job.setMapperClass(BasicWordCount.TokenizerMapper.class);

        job.setReducerClass(BasicWordCount.IntSumReducer.class);

        FileInputFormat.addInputPath(job, inDir);

        FileOutputFormat.setOutputPath(job, outDir);

        job.waitForCompletion(true);

        assertTrue(job.isSuccessful());

        // now check that the output is as expected

        List results = getJobResults(fs, outDir, 11);

        assertTrue(results.contains("She\t1"));

        assertTrue(results.contains("sells\t2"));

        // clean up after test case

        fs.delete(inDir, true);

        fs.delete(outDir, true);

    }

   /* @Test

    public void wordcount() throws Exception {

        HdfsConfiguration clusterConf = createSecureConfig("authentication,integrity,privacy");

        YarnConfiguration yarnConf =  createYarnSecureConfig();

        clusterConf.addResource(yarnConf);

        startCluster(clusterConf);

        Configuration conf = new Configuration();

        conf.addResource(FileUtils.openInputStream(new File("conf.Xml")));

        String IN_DIR = "testing/wordcount/input";

        String OUT_DIR = "testing/wordcount/output";

        String DATA_FILE = "sample.txt";

        FileSystem fs = FileSystem.get(conf);

        Path inDir = new Path(IN_DIR);

        Path outDir = new Path(OUT_DIR);

        fs.delete(inDir, true);

        fs.delete(outDir, true);

        // create the input data files

        List content = new ArrayList();

        content.add("She sells seashells at the seashore, and she sells nuts in the mountain.");

        writeHDFSContent(fs, inDir, DATA_FILE, content);

        String[] args = new String[]{IN_DIR,OUT_DIR};

        int exitCode = ToolRunner.run(conf,new WordCount(), args);

        fs.delete(inDir, true);

        fs.delete(outDir, true);

    }*/

    private void writeHDFSContent(FileSystem fs, Path dir, String fileName, List content) throws IOException {

        Path newFilePath = new Path(dir, fileName);

        FSDataOutputStream out = fs.create(newFilePath);

        for (String line : content) {

            out.writeBytes(line);

        }

        out.close();

    }

    protected List getJobResults(FileSystem fs, Path outDir, int numLines) throws Exception {

        List results = new ArrayList();

        FileStatus[] fileStatus = fs.listStatus(outDir);

        for (FileStatus file : fileStatus) {

            String name = file.getPath().getName();

            if (name.contains("part-r-00000")) {

                Path filePath = new Path(outDir + "/" + name);

                BufferedReader reader = new BufferedReader(new InputStreamReader(fs.open(filePath)));

                for (int i = 0; i < numLines; i++) {

                    String line = reader.readLine();

                    if (line == null) {

                        fail("Results are not what was expected");

                    }

                    System.out.println("line info: "+line);

                    results.add(line);

                }

                assertNull(reader.readLine());

                reader.close();

            }

        }

        return results;

    }

    private HdfsConfiguration createSecureConfig(String dataTransferProtection) throws Exception {

        HdfsConfiguration conf = new HdfsConfiguration();

        SecurityUtil.setAuthenticationMethod(UserGroupInformation.AuthenticationMethod.KERBEROS, conf);

        conf.set(DFS_NAMENODE_KERBEROS_PRINCIPAL_KEY, hdfsPrincipal);

        conf.set(DFS_NAMENODE_KEYTAB_FILE_KEY, keytab);

        conf.set(DFS_DATANODE_KERBEROS_PRINCIPAL_KEY, hdfsPrincipal);

        conf.set(DFS_DATANODE_KEYTAB_FILE_KEY, keytab);

        conf.set(DFS_WEB_AUTHENTICATION_KERBEROS_PRINCIPAL_KEY, spnegoPrincipal);

        conf.setBoolean(DFS_BLOCK_ACCESS_TOKEN_ENABLE_KEY, true);

        conf.set(DFS_DATA_TRANSFER_PROTECTION_KEY, dataTransferProtection);

        conf.set(DFS_HTTP_POLICY_KEY, HttpConfig.Policy.HTTPS_ONLY.name());

        conf.set(DFS_NAMENODE_HTTPS_ADDRESS_KEY, "localhost:0");

        conf.set(DFS_DATANODE_HTTPS_ADDRESS_KEY, "localhost:0");

        conf.setInt(IPC_CLIENT_CONNECT_MAX_RETRIES_ON_SASL_KEY, 10);

        conf.set(DFS_ENCRYPT_DATA_TRANSFER_KEY, "true");//https://issues.apache.org/jira/browse/HDFS-7431

        String keystoresDir = baseDir.getAbsolutePath();

        String sslConfDir = KeyStoreTestUtil.getClasspathDir(this.getClass());

        KeyStoreTestUtil.setupSSLConfig(keystoresDir, sslConfDir, conf, false);

        return conf;

    }

    private YarnConfiguration createYarnSecureConfig(){

        YarnConfiguration conf = new YarnConfiguration();

        //yarn secure config

        conf.set("yarn.resourcemanager.keytab", keytab);

        conf.set("yarn.resourcemanager.principal", hdfsPrincipal);

        conf.set("yarn.nodemanager.keytab", keytab);

        conf.set("yarn.nodemanager.principal", hdfsPrincipal);

        //   conf.set("yarn.nodemanager.container-executor.class", "org.apache.hadoop.yarn.server.nodemanager.DefaultContainerExecutor");

        conf.set("yarn.nodemanager.container-executor.class", "org.apache.hadoop.yarn.server.nodemanager.LinuxContainerExecutor");

        conf.set("yarn.nodemanager.linux-container-executor.path", "/container/container-executor");

        conf.set("mapreduce.jobhistory.keytab", keytab);

        conf.set("mapreduce.jobhistory.principal", hdfsPrincipal);

        conf.set("yarn.nodemanager.aux-services", "mapreduce_shuffle");//https://issues.apache.org/jira/browse/YARN-1289

        //enable security

        conf.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION, "true");

        //yarn

       conf.set("mapreduce.framework.name", "yarn");  //http://stackoverflow.com/questions/26567223/java-io-ioexception-cannot-initialize-cluster-in-hadoop2-with-yarn   use Yarn runner

        return conf;

    }

}

down vote

I have run into similar issues today. In my case I was building an über jar, where some dependency (I have not found the culprit yet) was bringing in a META-INF/services/org.apache.hadoop.mapreduce.protocol.ClientProtocolProvider with the contents:

org.apache.hadoop.mapred.LocalClientProtocolProvider

I provided my own in the project (e.g. put it on the classpath) with the following:

org.apache.hadoop.mapred.YarnClientProtocolProvider

and the correct one is picked up. I suspect you are seeing similar. To fix, please create the file described above, and put it on the classpath. If I find the culprit Jar, I will update the answer.

http://stackoverflow.com/questions/26567223/java-io-ioexception-cannot-initialize-cluster-in-hadoop2-with-yarn

hadoop-mapreduce-client-common-2.6.0.jar

#

#   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.

#

#org.apache.hadoop.mapred.LocalClientProtocolProvider

org.apache.hadoop.mapred.YarnClientProtocolProvider