我得到了这个奇怪的例外.我在Hadoop 2.6.4上使用Spark 1.6.0并在YARN集群上提交Spark作业.
16/07/23 20:05:21 WARN hdfs.DFSClient: DFSOutputStream ResponseProcessor exception for block BP-532134798-128.110.152.143-1469321545728:blk_1073741865_1041
java.io.EOFException: Premature EOF: no length prefix available
at org.apache.hadoop.hdfs.protocolPB.PBHelper.vintPrefixed(PBHelper.java:2203)
at org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck.readFields(PipelineAck.java:176)
at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer$ResponseProcessor.run(DFSOutputStream.java:867)
16/07/23 20:49:09 ERROR server.TransportRequestHandler: Error sending result RpcResponse{requestId=4719626006875125240, body=NioManagedBuffer{buf=java.nio.HeapByteBuffer[pos=0 lim=81 cap=81]}} to ms0440.utah.cloudlab.us/128.110.152.175:58944; closing connection
java.nio.channels.ClosedChannelException
在Hadoop 2.6.0运行时,我得到这个错误,并认为该例外可能是有点像一个错误的这个,但即使改变这对Hadoop的2.6.4后,我得到了同样的错误.没有任何内存问题,我的群集适用于HDFS和内存.我经历了这个和这个,但没有运气.
注意:1.我使用Apache Hadoop和Spark而不是任何CDH/HDP.2.我能够在HDFS中复制数据,甚至能够在该集群上执行另一个作业.
我有一些中间数据,我需要存储在HDFS和本地.我正在使用Spark 1.6.在HDFS作为中间形式我正在获取数据/output/testDummy/part-00000和/output/testDummy/part-00001.我想使用Java/Scala的地方保存这些分区,这样我可以将它们保存为/users/home/indexes/index.nt(双方在当地的合并),或者/users/home/indexes/index-0000.nt和/home/indexes/index-0001.nt分别.
这是我的代码:注意:testDummy与test相同,输出有两个分区.我想单独存储它们或组合它们但是本地存储index.nt文件.我更喜欢分别存储在两个数据节点中.我正在使用集群并在YARN上提交spark工作.我还添加了一些评论,多少次以及我得到的数据.我该怎么办？任何帮助表示赞赏.
val testDummy = outputFlatMapTuples.coalesce(Constants.INITIAL_PARTITIONS).saveAsTextFile(outputFilePathForHDFS+"/testDummy")
println("testDummy done") //1 time print
def savesData(iterator: Iterator[(String)]): Iterator[(String)] = {
println("Inside savesData") // now 4 times when coalesce(Constants.INITIAL_PARTITIONS)=2
println("iter size"+iterator.size) // 2 735 2 735 values
val filenamesWithExtension = outputPath + "/index.nt"
println("filenamesWithExtension "+filenamesWithExtension.length) //4 times
var list = List[(String)]()
val fileWritter = new FileWriter(filenamesWithExtension,true)
val bufferWritter = new BufferedWriter(fileWritter)
while (iterator.hasNext){ //iterator.hasNext is false
println("inside iterator") //0 times
val …我正在 Scala/Spark 中对行级别进行一些计算。我有一个使用下面的 JSON 创建的数据框 -
{"available":false,"createTime":"2016-01-08","dataValue":{"names_source":{"first_names":["abc", "def"],"last_names_id":[123,456]},"another_source_array":[{"first":"1.1","last":"ONE"}],"another_source":"TableSources","location":"GMP", "timestamp":"2018-02-11"},"deleteTime":"2016-01-08"}
您可以直接使用此 JSON 创建数据框。我的架构如下所示-
root
|-- available: boolean (nullable = true)
|-- createTime: string (nullable = true)
|-- dataValue: struct (nullable = true)
| |-- another_source: string (nullable = true)
| |-- another_source_array: array (nullable = true)
| | |-- element: struct (containsNull = true)
| | | |-- first: string (nullable = true)
| | | |-- last: string (nullable = true)
| |-- location: string (nullable = true)
| |-- …