A library for reading and writing Avro data from Spark SQL.
This documentation is for Spark 1.4+.
This library has different versions for Spark 1.2, 1.3, and 1.4+:
Spark Version | spark-avro version |
---|---|
1.2 |
0.2.0 |
1.3 |
1.0.0 |
1.4+ |
2.0.1 |
You can link against this library (for Spark 1.4+) in your program at the following coordinates:
Using SBT:
libraryDependencies += "com.databricks" %% "spark-avro" % "2.0.1"
Using Maven:
<dependency>
<groupId>com.databricks<groupId>
<artifactId>spark-avro_2.10</artifactId>
<version>2.0.1</version>
</dependency>
The spark-avro
library can also be added to Spark jobs launched through spark-shell
or spark-submit
by using the --packages
command line option.
For example, to include it when starting the spark shell:
$ bin/spark-shell --packages com.databricks:spark-avro_2.10:2.0.1
Unlike using --jars
, using --packages
ensures that this library and its dependencies will be added to the classpath.
The --packages
argument can also be used with bin/spark-submit
.
This library is cross-published for Scala 2.11, so 2.11 users should replace 2.10 with 2.11 in the commands listed above.
spark-avro
supports reading and writing of Avro data from Spark SQL.
- Automatic schema conversion:
spark-avro
supports most conversions between Spark SQL and Avro records, making Avro a first-class citizen in Spark. - Partitioning: This library allows developers to easily read and write partitioned data witout any extra configuration. Just pass the columns you want to partition on, just like you would for Parquet.
- Compression: You can specify the type of compression to use when writing Avro out to
disk. The supported types are
uncompressed
,snappy
, anddeflate
. You can also specify the deflate level. - Specifying record names: You can specify the record name and namespace to use by passing a map of parameters with
recordName
andrecordNamespace
.
spark-avro
supports reading all Avro types, with the exception of complex union
types. It uses the following mapping from Avro types to Spark SQL types:
Avro type | Spark SQL type |
---|---|
boolean | BooleanType |
int | IntegerType |
long | LongType |
float | FloatType |
double | DoubleType |
bytes | BinaryType |
string | StringType |
record | StructType |
enum | StringType |
array | ArrayType |
map | MapType |
fixed | BinaryType |
In addition to the types listed above, spark-avro
supports reading of three types of union
types:
union(int, long)
union(float, double)
union(something, null)
, wheresomething
is one of the supported Avro types listed above or is one of the supportedunion
types.
At the moment, spark-avro
ignores docs, aliases and other properties present in the Avro file.
spark-avro
supports writing of all Spark SQL types into Avro. For most types, the mapping from Spark types to Avro types is straightforward (e.g. IntegerType gets converted to int); however, there are a few special cases which are listed below:
Spark SQL type | Avro type |
---|---|
ByteType | int |
ShortType | int |
DecimalType | string |
BinaryType | bytes |
TimestampType | long |
StructType | record |
The recommended way to read or write Avro data from Spark SQL is by using Spark's DataFrame APIs, which are available in Scala, Java, Python, and R.
These examples use an Avro file available for download here:
// import needed for the .avro method to be added
import com.databricks.spark.avro._
val sqlContext = new SQLContext(sc)
// The Avro records get converted to Spark types, filtered, and
// then written back out as Avro records
val df = sqlContext.read.avro("src/test/resources/episodes.avro")
df.filter("doctor > 5").write.avro("/tmp/output")
Alternativly you can specify the format to use instead:
val sqlContext = new SQLContext(sc)
val df = sqlContext.read
.format("com.databricks.spark.avro")
.load("src/test/resources/episodes.avro")
df.filter("doctor > 5").write
.format("com.databricks.spark.avro")
.save("/tmp/output")
You can also specify Avro compression options:
import com.databricks.spark.avro._
val sqlContext = new SQLContext(sc)
// configuration to use deflate compression
sqlContext.setConf("spark.sql.avro.compression.codec", "deflate")
sqlContext.setConf("spark.sql.avro.deflate.level", "5")
val df = sqlContext.read.avro("src/test/resources/episodes.avro")
// writes out compressed Avro records
df.write.avro("/tmp/output")
You can write partitioned Avro records like this:
import com.databricks.spark.avro._
val sqlContext = new SQLContext(sc)
import sqlContext.implicits._
val df = Seq((2012, 8, "Batman", 9.8),
(2012, 8, "Hero", 8.7),
(2012, 7, "Robot", 5.5),
(2011, 7, "Git", 2.0))
.toDF("year", "month", "title", "rating")
df.write.partitionBy("year", "month").avro("/tmp/output")
You can specify the record name and namespace like this:
import com.databricks.spark.avro._
val sqlContext = new SQLContext(sc)
val df = sqlContext.read.avro("src/test/resources/episodes.avro")
val name = "AvroTest"
val namespace = "com.databricks.spark.avro"
val parameters = Map("recordName" -> name, "recordNamespace" -> namespace)
df.write.options(parameters).avro("/tmp/output")
import org.apache.spark.sql.*;
SQLContext sqlContext = new SQLContext(sc);
// Creates a DataFrame from a specified file
DataFrame df = sqlContext.read().format("com.databricks.spark.avro")
.load("src/test/resources/episodes.avro");
// Saves the subset of the Avro records read in
df.filter($"age > 5").write()
.format("com.databricks.spark.avro")
.save("/tmp/output");
# Creates a DataFrame from a specified directory
df = sqlContext.read.format("com.databricks.spark.avro").load("src/test/resources/episodes.avro")
# Saves the subset of the Avro records read in
subset = df.where("age > 5")
subset.write.format("com.databricks.spark.avro").save("output")
Avro data can be queried in pure SQL by registering the data as a temporary table.
CREATE TEMPORARY TABLE episodes
USING com.databricks.spark.avro
OPTIONS (path "src/test/resources/episodes.avro")
This library is built with SBT,
which is automatically downloaded by the included shell script. To build a JAR file simply run
build/sbt package
from the project root.
To run the tests, you should run build/sbt test
. In case you are doing improvements that target
speed, you can generate a sample Avro file and check how long it takes to read that Avro file
using the following commands:
build/sbt "test:run-main com.databricks.spark.avro.AvroFileGenerator NUMBER_OF_RECORDS NUMBER_OF_FILES"
will create sample avro files in target/avroForBenchmark/
. You can specify the number of records
for each file, as well as the overall number of files.
build/sbt "test:run-main com.databricks.spark.avro.AvroReadBenchmark"
runs count()
on the data inside target/avroForBenchmark/
and tells you how the operation took.
Similarly, you can do benchmarks on how long it takes to write DataFrame as Avro file with
build/sbt "test:run-main com.databricks.spark.avro.AvroWriteBenchmark NUMBER_OF_ROWS"
where NUMBER_OF_ROWS
is an optional parameter that allows you to specify the number of rows in DataFrame that we will be writing.