Skip to content
BranchesTags

Repository files navigation

CSV Data Source for Spark

A library for parsing and querying CSV data with Apache Spark, for Spark SQL and DataFrames.

Build Statuscodecov.io

Requirements

This library requires Spark 1.3+

Linking

You can link against this library in your program at the following coordinates:

Scala 2.10

groupId: com.databricks
artifactId: spark-csv_2.10
version: 1.3.0

Scala 2.11

groupId: com.databricks
artifactId: spark-csv_2.11
version: 1.3.0

Using with Spark shell

This package can be added to Spark using the --packages command line option. For example, to include it when starting the spark shell:

Spark compiled with Scala 2.11

$SPARK_HOME/bin/spark-shell --packages com.databricks:spark-csv_2.11:1.3.0

Spark compiled with Scala 2.10

$SPARK_HOME/bin/spark-shell --packages com.databricks:spark-csv_2.10:1.3.0

Features

This package allows reading CSV files in local or distributed filesystem as Spark DataFrames. When reading files the API accepts several options:

  • path: location of files. Similar to Spark can accept standard Hadoop globbing expressions.
  • header: when set to true the first line of files will be used to name columns and will not be included in data. All types will be assumed string. Default value is false.
  • delimiter: by default columns are delimited using ,, but delimiter can be set to any character
  • quote: by default the quote character is ", but can be set to any character. Delimiters inside quotes are ignored
  • escape: by default the escape character is \, but can be set to any character. Escaped quote characters are ignored
  • parserLib: by default it is "commons" can be set to "univocity" to use that library for CSV parsing.
  • mode: determines the parsing mode. By default it is PERMISSIVE. Possible values are:
    • PERMISSIVE: tries to parse all lines: nulls are inserted for missing tokens and extra tokens are ignored.
    • DROPMALFORMED: drops lines which have fewer or more tokens than expected or tokens which do not match the schema
    • FAILFAST: aborts with a RuntimeException if encounters any malformed line
  • charset: defaults to 'UTF-8' but can be set to other valid charset names
  • inferSchema: automatically infers column types. It requires one extra pass over the data and is false by default
  • comment: skip lines beginning with this character. Default is "#". Disable comments by setting this to null.
  • codec: compression codec to use when saving to file. Should be the fully qualified name of a class implementing org.apache.hadoop.io.compress.CompressionCodec or one of case-insensitive shorten names (bzip2, gzip, lz4, and snappy). Defaults to no compression when a codec is not specified.
  • nullValue: specificy a string that indicates a null value, any fields matching this string will be set as nulls in the DataFrame

The package also support saving simple (non-nested) DataFrame. When saving you can specify the delimiter and whether we should generate a header row for the table. See following examples for more details.

These examples use a CSV file available for download here:

$ wget https://github.com/databricks/spark-csv/raw/master/src/test/resources/cars.csv

SQL API

CSV data source for Spark can infer data types:

CREATETABLEcars
USING com.databricks.spark.csv
OPTIONS (path"cars.csv", header "true", inferSchema "true")

You can also specify column names and types in DDL.

CREATETABLEcars (yearMade double, carMake string, carModel string, comments string, blank string)
USING com.databricks.spark.csv
OPTIONS (path"cars.csv", header "true")

Scala API

Spark 1.4+:

Automatically infer schema (data types), otherwise everything is assumed string:

importorg.apache.spark.sql.SQLContextvalsqlContext=newSQLContext(sc)
valdf= sqlContext.read
.format("com.databricks.spark.csv")
.option("header", "true") // Use first line of all files as header
.option("inferSchema", "true") // Automatically infer data types
.load("cars.csv")
valselectedData= df.select("year", "model")
selectedData.write
.format("com.databricks.spark.csv")
.option("header", "true")
.save("newcars.csv")

You can manually specify the schema when reading data:

importorg.apache.spark.sql.SQLContextimportorg.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType};
valsqlContext=newSQLContext(sc)
valcustomSchema=StructType(
StructField("year", IntegerType, true),
StructField("make", StringType, true),
StructField("model", StringType, true),
StructField("comment", StringType, true),
StructField("blank", StringType, true))
valdf= sqlContext.read
.format("com.databricks.spark.csv")
.option("header", "true") // Use first line of all files as header
.schema(customSchema)
.load("cars.csv")
valselectedData= df.select("year", "model")
selectedData.write
.format("com.databricks.spark.csv")
.option("header", "true")
.save("newcars.csv")

You can save with compressed output:

importorg.apache.spark.sql.SQLContextvalsqlContext=newSQLContext(sc)
valdf= sqlContext.read
.format("com.databricks.spark.csv")
.option("header", "true") // Use first line of all files as header
.option("inferSchema", "true") // Automatically infer data types
.load("cars.csv")
valselectedData= df.select("year", "model")
selectedData.write
.format("com.databricks.spark.csv")
.option("header", "true")
.option("codec", "org.apache.hadoop.io.compress.GzipCodec")
.save("newcars.csv.gz")

Spark 1.3:

Automatically infer schema (data types), otherwise everything is assumed string:

importorg.apache.spark.sql.SQLContextvalsqlContext=newSQLContext(sc)
valdf= sqlContext.load(
"com.databricks.spark.csv",
Map("path"->"cars.csv", "header"->"true", "inferSchema"->"true"))
valselectedData= df.select("year", "model")
selectedData.save("newcars.csv", "com.databricks.spark.csv")

You can manually specify the schema when reading data:

importorg.apache.spark.sql.SQLContextimportorg.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType};
valsqlContext=newSQLContext(sc)
valcustomSchema=StructType(
StructField("year", IntegerType, true),
StructField("make", StringType, true),
StructField("model", StringType, true),
StructField("comment", StringType, true),
StructField("blank", StringType, true))
valdf= sqlContext.load(
"com.databricks.spark.csv",
schema = customSchema,
Map("path"->"cars.csv", "header"->"true"))
valselectedData= df.select("year", "model")
selectedData.save("newcars.csv", "com.databricks.spark.csv")

Java API

Spark 1.4+:

Automatically infer schema (data types), otherwise everything is assumed string:

importorg.apache.spark.sql.SQLContextSQLContextsqlContext = newSQLContext(sc);
DataFramedf = sqlContext.read()
.format("com.databricks.spark.csv")
.option("inferSchema", "true")
.option("header", "true")
.load("cars.csv");
df.select("year", "model").write()
.format("com.databricks.spark.csv")
.option("header", "true")
.save("newcars.csv");

You can manually specify schema:

importorg.apache.spark.sql.SQLContext;
importorg.apache.spark.sql.types.*;
SQLContextsqlContext = newSQLContext(sc);
StructTypecustomSchema = newStructType(newStructField[] {
newStructField("year", DataTypes.IntegerType, true, Metadata.empty()),
newStructField("make", DataTypes.StringType, true, Metadata.empty()),
newStructField("model", DataTypes.StringType, true, Metadata.empty()),
newStructField("comment", DataTypes.StringType, true, Metadata.empty()),
newStructField("blank", DataTypes.StringType, true, Metadata.empty())
});
DataFramedf = sqlContext.read()
.format("com.databricks.spark.csv")
.schema(customSchema)
.option("header", "true")
.load("cars.csv");
df.select("year", "model").write()
.format("com.databricks.spark.csv")
.option("header", "true")
.save("newcars.csv");

You can save with compressed output:

importorg.apache.spark.sql.SQLContextSQLContextsqlContext = newSQLContext(sc);
DataFramedf = sqlContext.read()
.format("com.databricks.spark.csv")
.option("inferSchema", "true")
.option("header", "true")
.load("cars.csv");
df.select("year", "model").write()
.format("com.databricks.spark.csv")
.option("header", "true")
.option("codec", "org.apache.hadoop.io.compress.GzipCodec")
.save("newcars.csv");

Spark 1.3:

Automatically infer schema (data types), otherwise everything is assumed string:

importorg.apache.spark.sql.SQLContextSQLContextsqlContext = newSQLContext(sc);
HashMap<String, String> options = newHashMap<String, String>();
options.put("header", "true");
options.put("path", "cars.csv");
options.put("inferSchema", "true");
DataFramedf = sqlContext.load("com.databricks.spark.csv", options);
df.select("year", "model").save("newcars.csv", "com.databricks.spark.csv");

You can manually specify schema:

importorg.apache.spark.sql.SQLContext;
importorg.apache.spark.sql.types.*;
SQLContextsqlContext = newSQLContext(sc);
StructTypecustomSchema = newStructType(newStructField[] {
newStructField("year", DataTypes.IntegerType, true, Metadata.empty()),
newStructField("make", DataTypes.StringType, true, Metadata.empty()),
newStructField("model", DataTypes.StringType, true, Metadata.empty()),
newStructField("comment", DataTypes.StringType, true, Metadata.empty()),
newStructField("blank", DataTypes.StringType, true, Metadata.empty())
});
HashMap<String, String> options = newHashMap<String, String>();
options.put("header", "true");
options.put("path", "cars.csv");
DataFramedf = sqlContext.load("com.databricks.spark.csv", customSchema, options);
df.select("year", "model").save("newcars.csv", "com.databricks.spark.csv");

You can save with compressed output:

importorg.apache.spark.sql.SQLContext;
importorg.apache.spark.sql.SaveMode;
SQLContextsqlContext = newSQLContext(sc);
HashMap<String, String> options = newHashMap<String, String>();
options.put("header", "true");
options.put("path", "cars.csv");
options.put("inferSchema", "true");
DataFramedf = sqlContext.load("com.databricks.spark.csv", options);
HashMap<String, String> saveOptions = newHashMap<String, String>();
saveOptions.put("header", "true");
saveOptions.put("path", "newcars.csv");
saveOptions.put("codec", "org.apache.hadoop.io.compress.GzipCodec");
df.select("year", "model").save("com.databricks.spark.csv", SaveMode.Overwrite,
saveOptions);

Python API

Spark 1.4+:

Automatically infer schema (data types), otherwise everything is assumed string:

frompyspark.sqlimportSQLContextsqlContext=SQLContext(sc)
df=sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('cars.csv')
df.select('year', 'model').write.format('com.databricks.spark.csv').save('newcars.csv')

You can manually specify schema:

frompyspark.sqlimportSQLContextfrompyspark.sql.typesimport*sqlContext=SQLContext(sc)
customSchema=StructType([ \
StructField("year", IntegerType(), True), \
StructField("make", StringType(), True), \
StructField("model", StringType(), True), \
StructField("comment", StringType(), True), \
StructField("blank", StringType(), True)])
df=sqlContext.read \
.format('com.databricks.spark.csv') \
.options(header='true') \
.load('cars.csv', schema=customSchema)
df.select('year', 'model').write \
.format('com.databricks.spark.csv') \
.save('newcars.csv')

You can save with compressed output:

frompyspark.sqlimportSQLContextsqlContext=SQLContext(sc)
df=sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('cars.csv')
df.select('year', 'model').write.format('com.databricks.spark.csv').options(codec="org.apache.hadoop.io.compress.GzipCodec").save('newcars.csv')

Spark 1.3:

Automatically infer schema (data types), otherwise everything is assumed string:

frompyspark.sqlimportSQLContextsqlContext=SQLContext(sc)
df=sqlContext.load(source="com.databricks.spark.csv", header='true', inferSchema='true', path='cars.csv')
df.select('year', 'model').save('newcars.csv', 'com.databricks.spark.csv')

You can manually specify schema:

frompyspark.sqlimportSQLContextfrompyspark.sql.typesimport*sqlContext=SQLContext(sc)
customSchema=StructType([ \
StructField("year", IntegerType(), True), \
StructField("make", StringType(), True), \
StructField("model", StringType(), True), \
StructField("comment", StringType(), True), \
StructField("blank", StringType(), True)])
df=sqlContext.load(source="com.databricks.spark.csv", header='true', schema=customSchema, path='cars.csv')
df.select('year', 'model').save('newcars.csv', 'com.databricks.spark.csv')

You can save with compressed output:

frompyspark.sqlimportSQLContextsqlContext=SQLContext(sc)
df=sqlContext.load(source="com.databricks.spark.csv", header='true', inferSchema='true', path='cars.csv')
df.select('year', 'model').save('newcars.csv', 'com.databricks.spark.csv', codec="org.apache.hadoop.io.compress.GzipCodec")

R API

Spark 1.4+:

Automatically infer schema (data types), otherwise everything is assumed string:

library(SparkR)
Sys.setenv('SPARKR_SUBMIT_ARGS'='"--packages" "com.databricks:spark-csv_2.10:1.3.0" "sparkr-shell"')
sqlContext<- sparkRSQL.init(sc)
df<- read.df(sqlContext, "cars.csv", source="com.databricks.spark.csv", inferSchema="true")
write.df(df, "newcars.csv", "com.databricks.spark.csv", "overwrite")

You can manually specify schema:

library(SparkR)
Sys.setenv('SPARKR_SUBMIT_ARGS'='"--packages" "com.databricks:spark-csv_2.10:1.3.0" "sparkr-shell"')
sqlContext<- sparkRSQL.init(sc)
customSchema<- structType(
structField("year", "integer"),
structField("make", "string"),
structField("model", "string"),
structField("comment", "string"),
structField("blank", "string"))
df<- read.df(sqlContext, "cars.csv", source="com.databricks.spark.csv", schema=customSchema)
write.df(df, "newcars.csv", "com.databricks.spark.csv", "overwrite")

You can save with compressed output:

library(SparkR)
Sys.setenv('SPARKR_SUBMIT_ARGS'='"--packages" "com.databricks:spark-csv_2.10:1.3.0" "sparkr-shell"')
sqlContext<- sparkRSQL.init(sc)
df<- read.df(sqlContext, "cars.csv", source="com.databricks.spark.csv", inferSchema="true")
write.df(df, "newcars.csv", "com.databricks.spark.csv", "overwrite", codec="org.apache.hadoop.io.compress.GzipCodec")

Building From Source

This library is built with SBT, which is automatically downloaded by the included shell script. To build a JAR file simply run sbt/sbt package from the project root. The build configuration includes support for both Scala 2.10 and 2.11.

About

CSV data source for Spark SQL and DataFrames

databricks.com/

Resources

Readme
Apache-2.0 license
Activity
Custom properties

Stars

0 stars

Watchers

17 watching

Forks

1 fork
Report repository

Releases

Packages

Contributors

Languages