forked from apache/spark
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[SPARK-1133] Add whole text files reader in MLlib
Here is a pointer to the former [PR164](apache#164). I add the pull request for the JIRA issue [SPARK-1133](https://spark-project.atlassian.net/browse/SPARK-1133), which brings a new files reader API in MLlib. Author: Xusen Yin <[email protected]> Closes apache#252 from yinxusen/whole-files-input and squashes the following commits: 7191be6 [Xusen Yin] refine comments 0af3faf [Xusen Yin] add JavaAPI test 01745ee [Xusen Yin] fix deletion error cc97dca [Xusen Yin] move whole text file API to Spark core d792cee [Xusen Yin] remove the typo character "+" 6bdf2c2 [Xusen Yin] test for small local file system block size a1f1e7e [Xusen Yin] add two extra spaces 28cb0fe [Xusen Yin] add whole text files reader
- Loading branch information
Showing
6 changed files
with
313 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
47 changes: 47 additions & 0 deletions
47
core/src/main/scala/org/apache/spark/input/WholeTextFileInputFormat.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You 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. | ||
| */ | ||
|
|
||
| package org.apache.spark.input | ||
|
|
||
| import org.apache.hadoop.fs.Path | ||
| import org.apache.hadoop.mapreduce.InputSplit | ||
| import org.apache.hadoop.mapreduce.JobContext | ||
| import org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat | ||
| import org.apache.hadoop.mapreduce.RecordReader | ||
| import org.apache.hadoop.mapreduce.TaskAttemptContext | ||
| import org.apache.hadoop.mapreduce.lib.input.CombineFileRecordReader | ||
| import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit | ||
|
|
||
| /** | ||
| * A [[org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat CombineFileInputFormat]] for | ||
| * reading whole text files. Each file is read as key-value pair, where the key is the file path and | ||
| * the value is the entire content of file. | ||
| */ | ||
|
|
||
| private[spark] class WholeTextFileInputFormat extends CombineFileInputFormat[String, String] { | ||
| override protected def isSplitable(context: JobContext, file: Path): Boolean = false | ||
|
|
||
| override def createRecordReader( | ||
| split: InputSplit, | ||
| context: TaskAttemptContext): RecordReader[String, String] = { | ||
|
|
||
| new CombineFileRecordReader[String, String]( | ||
| split.asInstanceOf[CombineFileSplit], | ||
| context, | ||
| classOf[WholeTextFileRecordReader]) | ||
| } | ||
| } |
72 changes: 72 additions & 0 deletions
72
core/src/main/scala/org/apache/spark/input/WholeTextFileRecordReader.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You 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. | ||
| */ | ||
|
|
||
| package org.apache.spark.input | ||
|
|
||
| import com.google.common.io.{ByteStreams, Closeables} | ||
|
|
||
| import org.apache.hadoop.io.Text | ||
| import org.apache.hadoop.mapreduce.InputSplit | ||
| import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit | ||
| import org.apache.hadoop.mapreduce.RecordReader | ||
| import org.apache.hadoop.mapreduce.TaskAttemptContext | ||
|
|
||
| /** | ||
| * A [[org.apache.hadoop.mapreduce.RecordReader RecordReader]] for reading a single whole text file | ||
| * out in a key-value pair, where the key is the file path and the value is the entire content of | ||
| * the file. | ||
| */ | ||
| private[spark] class WholeTextFileRecordReader( | ||
| split: CombineFileSplit, | ||
| context: TaskAttemptContext, | ||
| index: Integer) | ||
| extends RecordReader[String, String] { | ||
|
|
||
| private val path = split.getPath(index) | ||
| private val fs = path.getFileSystem(context.getConfiguration) | ||
|
|
||
| // True means the current file has been processed, then skip it. | ||
| private var processed = false | ||
|
|
||
| private val key = path.toString | ||
| private var value: String = null | ||
|
|
||
| override def initialize(split: InputSplit, context: TaskAttemptContext) = {} | ||
|
|
||
| override def close() = {} | ||
|
|
||
| override def getProgress = if (processed) 1.0f else 0.0f | ||
|
|
||
| override def getCurrentKey = key | ||
|
|
||
| override def getCurrentValue = value | ||
|
|
||
| override def nextKeyValue = { | ||
| if (!processed) { | ||
| val fileIn = fs.open(path) | ||
| val innerBuffer = ByteStreams.toByteArray(fileIn) | ||
|
|
||
| value = new Text(innerBuffer).toString | ||
| Closeables.close(fileIn, false) | ||
|
|
||
| processed = true | ||
| true | ||
| } else { | ||
| false | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
105 changes: 105 additions & 0 deletions
105
core/src/test/scala/org/apache/spark/input/WholeTextFileRecordReaderSuite.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,105 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You 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. | ||
| */ | ||
|
|
||
| package org.apache.spark.input | ||
|
|
||
| import java.io.DataOutputStream | ||
| import java.io.File | ||
| import java.io.FileOutputStream | ||
|
|
||
| import scala.collection.immutable.IndexedSeq | ||
|
|
||
| import com.google.common.io.Files | ||
|
|
||
| import org.scalatest.BeforeAndAfterAll | ||
| import org.scalatest.FunSuite | ||
|
|
||
| import org.apache.hadoop.io.Text | ||
|
|
||
| import org.apache.spark.SparkContext | ||
|
|
||
| /** | ||
| * Tests the correctness of | ||
| * [[org.apache.spark.input.WholeTextFileRecordReader WholeTextFileRecordReader]]. A temporary | ||
| * directory is created as fake input. Temporal storage would be deleted in the end. | ||
| */ | ||
| class WholeTextFileRecordReaderSuite extends FunSuite with BeforeAndAfterAll { | ||
| private var sc: SparkContext = _ | ||
|
|
||
| override def beforeAll() { | ||
| sc = new SparkContext("local", "test") | ||
|
|
||
| // Set the block size of local file system to test whether files are split right or not. | ||
| sc.hadoopConfiguration.setLong("fs.local.block.size", 32) | ||
| } | ||
|
|
||
| override def afterAll() { | ||
| sc.stop() | ||
| } | ||
|
|
||
| private def createNativeFile(inputDir: File, fileName: String, contents: Array[Byte]) = { | ||
| val out = new DataOutputStream(new FileOutputStream(s"${inputDir.toString}/$fileName")) | ||
| out.write(contents, 0, contents.length) | ||
| out.close() | ||
| } | ||
|
|
||
| /** | ||
| * This code will test the behaviors of WholeTextFileRecordReader based on local disk. There are | ||
| * three aspects to check: | ||
| * 1) Whether all files are read; | ||
| * 2) Whether paths are read correctly; | ||
| * 3) Does the contents be the same. | ||
| */ | ||
| test("Correctness of WholeTextFileRecordReader.") { | ||
|
|
||
| val dir = Files.createTempDir() | ||
| println(s"Local disk address is ${dir.toString}.") | ||
|
|
||
| WholeTextFileRecordReaderSuite.files.foreach { case (filename, contents) => | ||
| createNativeFile(dir, filename, contents) | ||
| } | ||
|
|
||
| val res = sc.wholeTextFiles(dir.toString).collect() | ||
|
|
||
| assert(res.size === WholeTextFileRecordReaderSuite.fileNames.size, | ||
| "Number of files read out does not fit with the actual value.") | ||
|
|
||
| for ((filename, contents) <- res) { | ||
| val shortName = filename.split('/').last | ||
| assert(WholeTextFileRecordReaderSuite.fileNames.contains(shortName), | ||
| s"Missing file name $filename.") | ||
| assert(contents === new Text(WholeTextFileRecordReaderSuite.files(shortName)).toString, | ||
| s"file $filename contents can not match.") | ||
| } | ||
|
|
||
| dir.delete() | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Files to be tested are defined here. | ||
| */ | ||
| object WholeTextFileRecordReaderSuite { | ||
| private val testWords: IndexedSeq[Byte] = "Spark is easy to use.\n".map(_.toByte) | ||
|
|
||
| private val fileNames = Array("part-00000", "part-00001", "part-00002") | ||
| private val fileLengths = Array(10, 100, 1000) | ||
|
|
||
| private val files = fileLengths.zip(fileNames).map { case (upperBound, filename) => | ||
| filename -> Stream.continually(testWords.toList.toStream).flatten.take(upperBound).toArray | ||
| }.toMap | ||
| } |