[HADOOP] 어떻게 자바에서 원격 원사 클러스터에 캐스 케이 딩 작업을 제출할 수 있습니까?

어떻게 자바에서 원격 원사 클러스터에 캐스 케이 딩 작업을 제출할 수 있습니까?

나는 캐스 케이 딩 사용 설명서에 설명 된대로 내가하는 JAR로 포장하여 캐스 케이 딩 작업을 제출할 수 있다는 것을 알고. 내가 수동으로 하둡 항아리 CLI 명령을 사용하여 제출하는 경우 그 작업은 내 클러스터에서 실행됩니다.

그러나, 원래의 하둡 1 캐스 케이 딩 버전, 하둡 JobConf의 특정 속성을 설정하여 클러스터에 작업을 제출하는 것이 가능했다. 설정 fs.defaultFS 및 mapred.job.tracker 로컬 하둡 라이브러리가 자동으로 Hadoop1은 JobTracker에 작업을 제출하려고 시도하는 원인이되었다. 그러나, 이러한 속성을 설정하면 최신 버전에서 작동하지 않습니다. 아래 설명과 같이, 서버와 협상 할 때 (지원 플랫폼 CDH5 나열) 계단식 버전 2.5.3를 사용하여 CDH5 5.2.1 하둡 클러스터에 제출하는 것은 IPC 예외로 이끈다.

2.5.6, 하둡 2, CDH 5, 원사 캐스 케이 딩, 및 제출에 대한 MR1의 API가 - -이 플랫폼 조합이 있다고 생각 호환성 테이블을 기반으로 지원되는 조합 (제목 "이전 자료"아래 참조). 하둡 항아리를 사용하여 작업을 제출하면이 같은 클러스터에서 잘 작동합니다. 포트 8031는 제출 호스트와 ResourceManager에 사이에 열려 있습니다. 같은 메시지와 함께 오류가 서버 측에서 ResourceManager에 로그에서 발견된다.

나는 계단식-hadoop2-MR1 라이브러리를 사용하고 있습니다.

Exception in thread "main" cascading.flow.FlowException: unhandled exception
    at cascading.flow.BaseFlow.complete(BaseFlow.java:894)
    at WordCount.main(WordCount.java:91)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at com.intellij.rt.execution.application.AppMain.main(AppMain.java:134)
Caused by: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.ipc.RpcServerException): Unknown rpc kind in rpc headerRPC_WRITABLE
    at org.apache.hadoop.ipc.Client.call(Client.java:1411)
    at org.apache.hadoop.ipc.Client.call(Client.java:1364)
    at org.apache.hadoop.ipc.WritableRpcEngine$Invoker.invoke(WritableRpcEngine.java:231)
    at org.apache.hadoop.mapred.$Proxy11.getStagingAreaDir(Unknown Source)
    at org.apache.hadoop.mapred.JobClient.getStagingAreaDir(JobClient.java:1368)
    at org.apache.hadoop.mapreduce.JobSubmissionFiles.getStagingDir(JobSubmissionFiles.java:102)
    at org.apache.hadoop.mapred.JobClient$2.run(JobClient.java:982)
    at org.apache.hadoop.mapred.JobClient$2.run(JobClient.java:976)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:415)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1614)
    at org.apache.hadoop.mapred.JobClient.submitJobInternal(JobClient.java:976)
    at org.apache.hadoop.mapred.JobClient.submitJob(JobClient.java:950)
    at cascading.flow.hadoop.planner.HadoopFlowStepJob.internalNonBlockingStart(HadoopFlowStepJob.java:105)
    at cascading.flow.planner.FlowStepJob.blockOnJob(FlowStepJob.java:196)
    at cascading.flow.planner.FlowStepJob.start(FlowStepJob.java:149)
    at cascading.flow.planner.FlowStepJob.call(FlowStepJob.java:124)
    at cascading.flow.planner.FlowStepJob.call(FlowStepJob.java:43)
    at java.util.concurrent.FutureTask.run(FutureTask.java:262)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)

데모 코드 계단식 사용자 가이드의 단어 수의 샘플 기본적으로 동일하다, 아래와 같다.

public class WordCount {

    public static void main(String[] args) {
        String inputPath = "/user/vagrant/wordcount/input";
        String outputPath = "/user/vagrant/wordcount/output";

        Scheme sourceScheme = new TextLine( new Fields( "line" ) );
        Tap source = new Hfs( sourceScheme, inputPath );

        Scheme sinkScheme = new TextDelimited( new Fields( "word", "count" ) );
        Tap sink = new Hfs( sinkScheme, outputPath, SinkMode.REPLACE );

        Pipe assembly = new Pipe( "wordcount" );


        String regex = "(?<!\\pL)(?=\\pL)[^ ]*(?<=\\pL)(?!\\pL)";
        Function function = new RegexGenerator( new Fields( "word" ), regex );
        assembly = new Each( assembly, new Fields( "line" ), function );


        assembly = new GroupBy( assembly, new Fields( "word" ) );

        Aggregator count = new Count( new Fields( "count" ) );
        assembly = new Every( assembly, count );

        Properties properties = AppProps.appProps()
            .setName( "word-count-application" )
            .setJarClass( WordCount.class )
            .buildProperties();

        properties.put("fs.defaultFS", "hdfs://192.168.30.101");
        properties.put("mapred.job.tracker", "192.168.30.101:8032");

        FlowConnector flowConnector = new HadoopFlowConnector( properties );
        Flow flow = flowConnector.connect( "word-count", source, sink, assembly );

        flow.complete();
    }
}

나는 또한 작업을 진행하려고 다른 속성의 무리를 설정 시도했다 :

이 중 어느 것도 그들은 단지 (mapred.job.tracker도 설정되어 있지 않으면) 작업이 로컬 모드에서 실행하게, 일을하지 않습니다.

해결법

1. ==============================

   1.지금은이 문제를 해결했습니다. 그것은 클라우 데라는, 특히 JobClient를 배포 이전 하둡 클래스를 사용하려고 시도에서 비롯됩니다. 당신이 제공 2.5.0-MR1 - cdh5.2.1 버전, 또는이 같은 버전 번호와 하둡 클라이언트 의존성 하둡 코어를 사용하는 경우이 오류가 발생할 것입니다. 이것은 MR1 버전 주장, 우리가 제출 MR1 API를 사용하고 있지만,이 버전은 실제로 만 Hadoop1은 JobTracker에 제출을 지원하며, 원사를 지원하지 않습니다.

   지금은이 문제를 해결했습니다. 그것은 클라우 데라는, 특히 JobClient를 배포 이전 하둡 클래스를 사용하려고 시도에서 비롯됩니다. 당신이 제공 2.5.0-MR1 - cdh5.2.1 버전, 또는이 같은 버전 번호와 하둡 클라이언트 의존성 하둡 코어를 사용하는 경우이 오류가 발생할 것입니다. 이것은 MR1 버전 주장, 우리가 제출 MR1 API를 사용하고 있지만,이 버전은 실제로 만 Hadoop1은 JobTracker에 제출을 지원하며, 원사를 지원하지 않습니다.

   원사에 제출을 허용하기 위해, 당신은 여전히 ​​원사에 MR1 작업의 제출을 ​​지원하는 비 MR1 2.5.0-cdh5.2.1 버전과 하둡 클라이언트 의존성을 사용해야합니다.

from https://stackoverflow.com/questions/27606807/how-can-i-submit-a-cascading-job-to-a-remote-yarn-cluster-from-java by cc-by-sa and MIT license