Hadoop Pipes Exception: Illegal text protocol command

Hadoop Pipes Exception: Illegal text protocol command

对于Hadoop pipes 出现这样的错误,基本上编译代码依赖的.so和.a 版本不匹配

网上也没有给出更多信息,我的同事最近回复了解决办法,可以参考

https://groups.google.com/a/cloudera.org/forum/#!msg/cdh-user/j0dpYPDx3_A/S1rotrWGSf0J

如果不能翻墙,摘录如下

 

I met with the same issue, and my way to work around it is to apply the attached patch, regenerate the native libraries(libhadooppipes.a, libhadooputils.a, etc.) and recompile user program with the new libraries. Notice that this is only for running against MR1, and don't mess up with those against Yarn. Hope this also works in your case. :-)

 

p.s. I used "mvn -Dskiptests -Pdist,native -Dtar clean package" cmd to generate the tarball and got the native libs under "lib/native" dir

Index: hadoop-tools/hadoop-pipes/src/main/native/pipes/impl/HadoopPipes.cc
===================================================================
--- hadoop-tools/hadoop-pipes/src/main/native/pipes/impl/HadoopPipes.cc    (revision 340277)
+++ hadoop-tools/hadoop-pipes/src/main/native/pipes/impl/HadoopPipes.cc    (working copy)
@@ -30,7 +30,6 @@
 #include <stdlib.h>
 #include <string.h>
 #include <strings.h>
-#include <unistd.h>
 #include <sys/socket.h>
 #include <pthread.h>
 #include <iostream>
@@ -806,8 +805,8 @@
       }
       if (reducer != NULL) {
         int64_t spillSize = 100;
-        if (jobConf->hasKey("mapreduce.task.io.sort.mb")) {
-          spillSize = jobConf->getInt("mapreduce.task.io.sort.mb");
+        if (jobConf->hasKey("io.sort.mb")) {
+          spillSize = jobConf->getInt("io.sort.mb");
         }
         writer = new CombineRunner(spillSize * 1024 * 1024, this, reducer, 
                                    uplink, partitioner, numReduces);
@@ -1042,7 +1041,7 @@
    */
   void* ping(void* ptr) {
     TaskContextImpl* context = (TaskContextImpl*) ptr;
-    char* portStr = getenv("mapreduce.pipes.command.port");
+    char* portStr = getenv("hadoop.pipes.command.port");
     int MAX_RETRIES = 3;
     int remaining_retries = MAX_RETRIES;
     while (!context->isDone()) {
@@ -1095,7 +1094,7 @@
     try {
       TaskContextImpl* context = new TaskContextImpl(factory);
       Protocol* connection;
-      char* portStr = getenv("mapreduce.pipes.command.port");
+      char* portStr = getenv("hadoop.pipes.command.port");
       int sock = -1;
       FILE* stream = NULL;
       FILE* outStream = NULL;
@@ -1128,8 +1127,8 @@
         HADOOP_ASSERT(setbuf == 0, string("problem with setvbuf for outStream: ")
                                      + strerror(errno));
         connection = new BinaryProtocol(stream, context, outStream);
-      } else if (getenv("mapreduce.pipes.commandfile")) {
-        char* filename = getenv("mapreduce.pipes.commandfile");
+      } else if (getenv("hadoop.pipes.command.file")) {
+        char* filename = getenv("hadoop.pipes.command.file");
         string outFilename = filename;
         outFilename += ".out";
         stream = fopen(filename, "r");