Load CSV file with PySpark

by

Spark 2.0.0+

You can use built-in csv data source directly:

spark.read.csv(
    "some_input_file.csv", 
    header=True, 
    mode="DROPMALFORMED", 
    schema=schema
)

or

(
    spark.read
    .schema(schema)
    .option("header", "true")
    .option("mode", "DROPMALFORMED")
    .csv("some_input_file.csv")
)

without including any external dependencies.

Spark < 2.0.0:

Instead of manual parsing, which is far from trivial in a general case, I would recommend spark-csv:

Make sure that Spark CSV is included in the path (--packages, --jars, --driver-class-path)

And load your data as follows:

df = (
    sqlContext
    .read.format("com.databricks.spark.csv")
    .option("header", "true")
    .option("inferschema", "true")
    .option("mode", "DROPMALFORMED")
    .load("some_input_file.csv")
)

It can handle loading, schema inference, dropping malformed lines and doesn’t require passing data from Python to the JVM.

Note:

If you know the schema, it is better to avoid schema inference and pass it to DataFrameReader. Assuming you have three columns – integer, double and string:

from pyspark.sql.types import StructType, StructField
from pyspark.sql.types import DoubleType, IntegerType, StringType

schema = StructType([
    StructField("A", IntegerType()),
    StructField("B", DoubleType()),
    StructField("C", StringType())
])

(
    sqlContext
    .read
    .format("com.databricks.spark.csv")
    .schema(schema)
    .option("header", "true")
    .option("mode", "DROPMALFORMED")
    .load("some_input_file.csv")
)

More Related Contents:

  1. Load CSV file with Spark
  2. Spark Equivalent of IF Then ELSE
  3. How to add a constant column in a Spark DataFrame?
  4. How do I add a new column to a Spark DataFrame (using PySpark)?
  5. How to perform union on two DataFrames with different amounts of columns in spark?
  6. Does spark predicate pushdown work with JDBC?
  7. ‘PipelinedRDD’ object has no attribute ‘toDF’ in PySpark
  8. Count number of non-NaN entries in each column of Spark dataframe with Pyspark
  9. Explode in PySpark
  10. Spark DataFrame: Computing row-wise mean (or any aggregate operation)
  11. Passing a data frame column and external list to udf under withColumn
  12. Updating a dataframe column in spark
  13. PySpark converting a column of type ‘map’ to multiple columns in a dataframe
  14. Spark add new column to dataframe with value from previous row
  15. Best way to get the max value in a Spark dataframe column
  16. Pyspark: Replacing value in a column by searching a dictionary
  17. Apache Spark Python Cosine Similarity over DataFrames
  18. Pivot String column on Pyspark Dataframe
  19. GroupBy column and filter rows with maximum value in Pyspark
  20. Using UDF ignores condition in when
  21. Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
  22. Concatenate two PySpark dataframes
  23. How do I convert an array (i.e. list) column to Vector
  24. Filtering DataFrame using the length of a column
  25. Add column sum as new column in PySpark dataframe
  26. How to add suffix and prefix to all columns in python/pyspark dataframe
  27. How to drop all columns with null values in a PySpark DataFrame?
  28. Pyspark changing type of column from date to string
  29. PySpark: compute row maximum of the subset of columns and add to an exisiting dataframe
  30. how to add Row id in pySpark dataframes [duplicate]

Leave a Comment