将文件写入 Hdfs 时出现 IOException答案

【问题标题】：IOException on writing a file to Hdfs将文件写入 Hdfs 时出现 IOException
【发布时间】：2017-12-07 01:53:44
【问题描述】：

我正在开发一个程序，该程序从 mqtt 代理获取 gps 数据并将其加载到 hadoop 集群中。在尝试将数据写入 hdfs 时，我得到了 IOException。以下是完整的堆栈跟踪：

java.io.IOException: Failed on local exception: com.google.protobuf.InvalidProtocolBufferException: Message missing required fields: callId, status; Host Details : local host is: "quickstart.cloudera/192.168.25.170"; destination host is: "quickstart.cloudera":8020; 
    at org.apache.hadoop.net.NetUtils.wrapException(NetUtils.java:765)
    at org.apache.hadoop.ipc.Client.call(Client.java:1165)
    at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:184)
    at com.sun.proxy.$Proxy7.create(Unknown Source)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:165)
    at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:84)
    at com.sun.proxy.$Proxy7.create(Unknown Source)
    at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.create(ClientNamenodeProtocolTranslatorPB.java:187)
    at org.apache.hadoop.hdfs.DFSOutputStream.<init>(DFSOutputStream.java:1250)
    at org.apache.hadoop.hdfs.DFSOutputStream.newStreamForCreate(DFSOutputStream.java:1269)
    at org.apache.hadoop.hdfs.DFSClient.create(DFSClient.java:1063)
    at org.apache.hadoop.hdfs.DFSClient.create(DFSClient.java:1021)
    at org.apache.hadoop.hdfs.DistributedFileSystem.create(DistributedFileSystem.java:232)
    at org.apache.hadoop.hdfs.DistributedFileSystem.create(DistributedFileSystem.java:75)
    at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:806)
    at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:787)
    at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:686)
    at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:675)
    at com.mqttHadoopLoader.hadoop.MqttLoader.HdfsWriter.writeToHdfs(HdfsWriter.java:19)
    at com.mqttHadoopLoader.hadoop.MqttLoader.MqttDataLoader.messageArrived(MqttDataLoader.java:43)
    at org.eclipse.paho.client.mqttv3.internal.CommsCallback.handleMessage(CommsCallback.java:354)
    at org.eclipse.paho.client.mqttv3.internal.CommsCallback.run(CommsCallback.java:162)
    at java.lang.Thread.run(Thread.java:748)
Caused by: com.google.protobuf.InvalidProtocolBufferException: Message missing required fields: callId, status
    at com.google.protobuf.UninitializedMessageException.asInvalidProtocolBufferException(UninitializedMessageException.java:81)
    at org.apache.hadoop.ipc.protobuf.RpcPayloadHeaderProtos$RpcResponseHeaderProto$Builder.buildParsed(RpcPayloadHeaderProtos.java:1094)
    at org.apache.hadoop.ipc.protobuf.RpcPayloadHeaderProtos$RpcResponseHeaderProto$Builder.access$1300(RpcPayloadHeaderProtos.java:1028)
    at org.apache.hadoop.ipc.protobuf.RpcPayloadHeaderProtos$RpcResponseHeaderProto.parseDelimitedFrom(RpcPayloadHeaderProtos.java:986)
    at org.apache.hadoop.ipc.Client$Connection.receiveResponse(Client.java:850)
    at org.apache.hadoop.ipc.Client$Connection.run(Client.java:781)

当我尝试创建 OutputStream 时似乎发生了错误，但很难说，因为我的 eclipse 调试器工作不正常（它说它无法连接到 VM，我已经尝试了多种修复在stackoverflow上发布）。这是我的 HdfsWriter 代码：

String destFile = "hdfs://127.0.0.0.1:8020/gpsData/output/gps_data.txt";
 //Note *this is just a placeholder IP address for the purpose of this post. I do have the fully correct IP address for the program. 

    public void writeToHdfs(String gpsInfo) {
        try {
            Configuration conf = new Configuration();
            System.out.println("Connecting to -- " + conf.get("fs.defaultFS"));

            FileSystem fs = FileSystem.get(URI.create(destFile), conf);
            System.out.println(fs.getUri());

            // Error seems to occur here
            OutputStream outStream = fs.create(new Path(destFile));

            byte[] messageByt = gpsInfo.getBytes();
            outStream.write(messageByt);
            outStream.close();

            System.out.println(destFile + " copied to HDFS");

        } catch (FileNotFoundException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

这是调用 HdfsWriter 的 mqtt 方法：

public void messageArrived(String topic, MqttMessage message)
        throws Exception {
            System.out.println(message);
            HdfsWriter hdfsWriter = new HdfsWriter();
            hdfsWriter.writeToHdfs(message.toString());
    }

我对 hadoop 还是很陌生，所以任何和所有的帮助都会很棒。

更新我已经进行了调试，并且可以明确地告诉您，每当我尝试调用 FileSystem 方法时都会发生错误。例如，fs.exists(pt) 和fs.setReplication() 也会触发错误。

【问题讨论】：

您需要仔细阅读错误信息。它会告诉您出了什么问题：InvalidProtocolBufferException: Message missing required fields: callId, status;。因此，您正在向服务器发送无效数据。问题在于您发送的内容，而不是发送方式。
@JB Nizet 是的，我已阅读错误消息。多次。一遍又一遍地。我什至不知道错误消息在说什么“消息”。如果您查看我包含的代码，您会发现我没有任何东西可以直接调用任何名为“消息”的东西。我的调试工作正常，可以告诉您，我尝试调用的任何 FileSystem 消息都会发生这种情况（即，fs.exists(pt) 或 fs.setReplication(pt, (short)1) 也会发生这种情况）。我找不到这个错误直接发生在哪里。

标签： java hadoop ioexception

【解决方案1】：

我相信使用 google protobuf 库的 hdfs。而且您的客户端代码似乎使用了错误（不兼容）版本的 protobuf。尝试朝这个方向挖掘。

【讨论】：

【解决方案2】：

HDFS 客户端和 NameNode 之间的协议使用 Google Protocol Buffers 来序列化消息。该错误表示客户端发送的消息未包含所有预期的字段，因此与服务器不兼容。

这可能表明您正在运行的 HDFS 客户端版本比 NameNode 版本旧。例如，callId 字段在 Apache JIRA 问题HADOOP-9762 跟踪的功能中实现，并在 Apache Hadoop 2.1.0-beta 中提供。该版本之前的客户端不会在其消息中包含callId，因此它将与运行 2.1.0-beta 或更高版本的 NameNode 不兼容。

我建议检查您的客户端应用程序，以确保它使用的 Hadoop 客户端库版本与 Hadoop 集群版本相匹配。从堆栈跟踪来看，您似乎正在使用 Cloudera 发行版。如果是这样，那么通过使用 Cloudera 在其 Maven 存储库中提供的匹配客户端库依赖项版本，您可能会获得最大的成功。详情请见Using the CDH 5 Maven Repository。

【讨论】：