Spark - How to read CSV File

-

In this post, I would like to show how to read a CSV file located on your local file system.

First, lets consider the below CSV file. You can write it using Excel, Libra Office Calc or any simple text editor. Note, we have 6 columns - A to F and 5 rows where the first row is a header row.

 

Name,Age,Birth Date,Id,Address,Score
Arike,28,1991-02-14,3698547891,"New York, NY 10022, USA",78.6
Bob,32,1987-06-07,6984184782,"1843-1701 S Osage Dr, Ottawa, KS 66067, USA",45.32
Corry,65,1954-12-26,9782472174,"R. Olavo Bilac, 143 - Batel, Curitiba - PR, 80440-040, Brazil",98.47
David,18,2001-10-22,2316324177,"20-16 B5036, Wirksworth, Matlock DE4 4FP, UK",3.77

Assuming we have an initialized SparkSession object called spark we can write the following.
Here, the "read" property returns a DataFrameReader we can execute the csv method with a path to the csv.
The line df.printSchema() prints to the console the scheme of the file.
df = spark.read\
    .csv('data.csv')
df.printSchema()

df.show()

And the output will be:
root
 |-- _c0: string (nullable = true)
 |-- _c1: string (nullable = true)
 |-- _c2: string (nullable = true)
 |-- _c3: string (nullable = true)
 |-- _c4: string (nullable = true)
 |-- _c5: string (nullable = true)



+-----+---+----------+----------+--------------------+-----+
|  _c0|_c1|       _c2|       _c3|                 _c4|  _c5|
+-----+---+----------+----------+--------------------+-----+
| Name|Age|Birth Date|        Id|             Address|Score|
|Arike| 28|1991-02-14|3698547891|New York, NY 1002...| 78.6|
|  Bob| 32|1987-06-07|6984184782|1843-1701 S Osage...|45.32|
|Corry| 65|1954-12-26|9782472174|R. Olavo Bilac, 1...|98.47|
|David| 18|2001-10-22|2316324177|20-16 B5036, Wirk...| 3.77|
+-----+---+----------+----------+--------------------+-----+

Note that the column names are _c0, _c1 etc. and all the types are strings. So let fix that.

df = spark.read\
    .csv('data.csv', header=True, inferSchema=True)

df.printSchema()
df.show()
The optional input "header" is set to True so the Spark will refer to the first line as header names otherwise Spark will generate its own names like _c0, _c1 etc. The optional input "inferSchema" is set to True so Spark will sample the data and try to guess its type. Lets see the output:
root
 |-- Name: string (nullable = true)
 |-- Age: integer (nullable = true)
 |-- Birth Date: timestamp (nullable = true)
 |-- Id: long (nullable = true)
 |-- Address: string (nullable = true)
 |-- Score: double (nullable = true)

+-----+---+-------------------+----------+--------------------+-----+
| Name|Age|         Birth Date|        Id|             Address|Score|
+-----+---+-------------------+----------+--------------------+-----+
|Arike| 28|1991-02-14 00:00:00|3698547891|New York, NY 1002...| 78.6|
|  Bob| 32|1987-06-07 00:00:00|6984184782|1843-1701 S Osage...|45.32|
|Corry| 65|1954-12-26 00:00:00|9782472174|R. Olavo Bilac, 1...|98.47|
|David| 18|2001-10-22 00:00:00|2316324177|20-16 B5036, Wirk...| 3.77|
+-----+---+-------------------+----------+--------------------+-----+

Alternatively, we could specify the schema ourselves. In the example below I remove the inferSchema input, and the default value of inferSchema is False. Additionally, I specify that the Id column is a string not long as before.

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

schema = StructType([
    StructField("Name", StringType(), True),
    StructField("Age", IntegerType(), True),
    StructField("Birth Date", TimestampType(), True),
    StructField("Id", StringType(), True),
    StructField("Address", StringType(), True),
    StructField("Score", DoubleType(), True)
])
df = spark.read\
    .csv('data.csv', header=True, schema=schema)

df.printSchema()
df.show()



root
 |-- Name: string (nullable = true)
 |-- Age: integer (nullable = true)
 |-- Birth Date: timestamp (nullable = true)
 |-- Id: string (nullable = true)
 |-- Address: string (nullable = true)
 |-- Score: double (nullable = true)

+-----+---+-------------------+----------+--------------------+-----+
| Name|Age|         Birth Date|        Id|             Address|Score|
+-----+---+-------------------+----------+--------------------+-----+
|Arike| 28|1991-02-14 00:00:00|3698547891|New York, NY 1002...| 78.6|
|  Bob| 32|1987-06-07 00:00:00|6984184782|1843-1701 S Osage...|45.32|
|Corry| 65|1954-12-26 00:00:00|9782472174|R. Olavo Bilac, 1...|98.47|
|David| 18|2001-10-22 00:00:00|2316324177|20-16 B5036, Wirk...| 3.77|
+-----+---+-------------------+----------+--------------------+-----+

Of course, there are more parameters to this API, I will leave it up to you to explore it in this link to Spark documentation.



Comments

Post a Comment

Popular posts from this blog

5 ways to calculate max of row in Spark

-
Image
5 ways to calculate max of row in Spark/PySpark In this post I tested the performance of 5 different methods to calculate a maximal value of a single row. You might encounter this situation before: you have multiple columns with some value, integer or float for example, and you want to calculate the maximum value per row for all the data frame and append it as additional column. We will test 5 different ways to calculate maximum value: udf - user define function 'array' and 'array_max' functions 'greatest' function rdd 'map' transformation data frame groupBy and max-aggregation For each of these methods will count the time it take to read a CSV file, calculate the maximal value and add it as 'max' column. For the data, we are going to generate 10 columns of integer values over 1 million rows via Python code. The values will be in the range of 1 to 1000. import random def gen_csv (): f = open ( "../data_10x1m.csv&quo
Read more

Spark - How to read a JSON file

-
Image
First thing to notice it the file structure. Its actually a collection of json objects *each one in a single line*. { "field_name": "value", "field_name": "value", ... } { "field_name": "value", "field_name": "value", ... } { "field_name": "value", "field_name": "value", ... } Unlike csv() method the json() method does not have inferSchema parameter. The schema is inferred from the json types. from pyspark.sql import SparkSession import pyspark.sql.types as t import pyspark.sql.functions as f spark = SparkSession . builder \ . master( 'local' ) \ . appName( 'Read JSON file' ) \ . getOrCreate() df = spark . read \ . json( 'data.json' , multiLine = True ) df . printSchema() df . show(truncate = False ) root |-- Address: string (nullable = true) |-- Age: long (nullable = true) |-- Birth Date: string (nu
Read more

Create Custom Datasource With Spark 3 - Part 1

-
Image
This article is the first of a series on how to create a custom data source that will plug in to Spark mechanism. The idea is to write something like: Dataset < Row > df = sparkSession . read () . option ( "filename" , "path-to-data-file" ) . option ( "other-options" , "some value" ) . format ( "com.roizaig.mydatasource" ) . load (); To read a file, parse it according to some custom logic and return a dataset that represents it. Why Create Custom Data Source? Spark support out of the box in Parquet , CSV , JSON and ORC  and more. In most cases these formats will be more than okay and if not, you can always read it as a text file and do some string manipulation in Spark. However, there are some cases where we would like to "teach" Spark to read our own custom format. Flexibility is a key point, with a custom datasource you can read every format y
Read more