Skip to content
This repository was archived by the owner on Dec 20, 2018. It is now read-only.

Pick last file sorted by path for schema #269

Open
koertkuipers wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Pick last file sorted by path for schema #269

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
a30d700
make the avro file picked for schema stable by sorting on path and pi…
koertkuipers May 2, 2017
bdefef8
test reading files with different schemas and picking the schema cons…
koertkuipers Feb 12, 2018
4867bcf
use min for picking first path
koertkuipers Jun 3, 2018
dc6dcdd
add one more test for predictably picking schema
koertkuipers Jun 3, 2018
64c10b1
can't stand scalastyle
koertkuipers Jun 3, 2018
c82d354
pick last file instead of first
koertkuipers Jun 7, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 16 additions & 12 deletions src/main/scala/com/databricks/spark/avro/DefaultSource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,24 +61,28 @@ private[avro] class DefaultSource extends FileFormat with DataSourceRegister {
files: Seq[FileStatus]): Option[StructType] = {
val conf = spark.sparkContext.hadoopConfiguration

// Schema evolution is not supported yet. Here we only pick a single random sample file to
// Schema evolution is not supported yet. Here we only pick the first file sorted by path to
// figure out the schema of the whole dataset.
val sampleFile = if (conf.getBoolean(IgnoreFilesWithoutExtensionProperty, true)) {
files.find(_.getPath.getName.endsWith(".avro")).getOrElse {
throw new FileNotFoundException(
"No Avro files found. Hadoop option \"avro.mapred.ignore.inputs.without.extension\" is " +
"set to true. Do all input files have \".avro\" extension?"
)
}
def sampleFilePath = if (conf.getBoolean(IgnoreFilesWithoutExtensionProperty, true)) {
files.iterator.map(_.getPath).filter(_.getName.endsWith(".avro"))

@gengliangwang gengliangwang Jun 1, 2018 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

files.map(.getPath).sortBy(.getName)....

@koertkuipers koertkuipers Jun 1, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it has same result right?

files can be a very large sequence. the iterator approach avoids creating 2 copies of that sequence. also it is not necessary to do a full sort just to get the first sorted element.

are you saying its not worth the optimization?

@gengliangwang gengliangwang Jun 1, 2018 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are right for not sorting all the file names.
But I don't think we need to convert it to an iterator.
Maybe we can try to make it more shorter like files.map(_.getPath).minBy(_.getName) ?
We can create a function which accepts parameter Seq(Path), then check if it is empty before getting the minimal one.

@koertkuipers koertkuipers Jun 3, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

iterator is lightweight and avoids materialization

minBy(_.getName) wouldnt work because we want to sort by the path, not just the filename (e.g. /some/path/x=1/part-0000.avro comes before /some/path/x=2/part-0000.avro)

minBy(_.toString) might work but i don't feel too certain about it. rather use Comparable to do the right thing. unfortunately Path is just Comparable, not Comparable[Path], so scala doesn't understand how to use it, which is why i resorted to using compareTo directly.

.reduceOption{ (p1, p2) => if (p1.compareTo(p2) <= 0) p1 else p2 }
.getOrElse {
throw new FileNotFoundException(
"No Avro files found. Hadoop option \"avro.mapred.ignore.inputs.without.extension\" " +
"is set to true. Do all input files have \".avro\" extension?"
)
}
} else {
files.headOption.getOrElse {
throw new FileNotFoundException("No Avro files found.")
}
files.iterator.map(_.getPath)

@gengliangwang gengliangwang Jun 1, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ditto

.reduceOption{ (p1, p2) => if (p1.compareTo(p2) <= 0) p1 else p2 }
.getOrElse{
throw new FileNotFoundException("No Avro files found.")
}
}

// User can specify an optional avro json schema.
val avroSchema = options.get(AvroSchema).map(new Schema.Parser().parse).getOrElse {
val in = new FsInput(sampleFile.getPath, conf)
val in = new FsInput(sampleFilePath, conf)
try {
val reader = DataFileReader.openReader(in, new GenericDatumReader[GenericRecord]())
try {
Expand Down
11 changes: 11 additions & 0 deletions src/test/scala/com/databricks/spark/avro/AvroSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -803,4 +803,15 @@ class AvroSuite extends FunSuite with BeforeAndAfterAll {
assert(readDf.collect().sameElements(writeDf.collect()))
}
}

test("writing avro partitions with different schemas and reading back out with a single predictable schema") {
TestUtils.withTempDir { tempDir =>
val df1 = spark.createDataFrame(Seq(("a", 1), ("b", 2)))
df1.write.avro(s"$tempDir/different_schemas/z=1")
val df2 = spark.createDataFrame(Seq(Tuple1("a"), Tuple1("b")))
df2.write.avro(s"$tempDir/different_schemas/z=2")
val df3 = spark.read.avro(s"$tempDir/different_schemas")

@gengliangwang gengliangwang Jun 1, 2018 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

maybe add a loop for the reading? I am not sure if the order will be different every time

assert(df3.schema.fieldNames.toSet === Set("_1", "_2", "z"))
}
}
}