Spark - How to create a DataFrame (different ways)

-

How to create a Spark DataFrame in different ways



DataFrames are a table-like interface in Spark. The DataFrame structure is a distributed collection of data, build on Dataset. Since DataFrame have a defined schema and types it allows Spark to utilize different optimization on top of the schema and to execute calculations faster. Below we have initial code to instantiate SparkContext, SQLContext and SparkSession code. 

from pyspark.sql import Row
from pyspark import SparkContext
from pyspark import SQLContext
sc = SparkContext()
sqlContext = SQLContext(sc)
from pyspark.sql import SparkSession
spark = SparkSession.builder \
    .master('local') \
    .appName('Create DataFrame') \
    .getOrCreate()

toDF method

Perhaps the most simple way to create a DataFrame is to use the toDF method. All you need is to supply a list of column names (strings) and a list of tuples as data. If you don't specify the column names Spark generate the column names, like _1, _2, etc. 

list_of_persons = [('Arike', 28, 78.6), ('Bob', 32, 45.32), ('Corry', 65, 98.47)]
df = sc.parallelize(list_of_persons).toDF(['name', 'age', 'score'])
df.printSchema()
df.show()

The parallelize method returns an RDD which then converted to DataFrame. Note that the column types are inferred from the data. Also, the columns are nullable = true so null's or None's are permitted even if you wish otherwise. 

root
 |-- name: string (nullable = true)
 |-- age: long (nullable = true)
 |-- score: double (nullable = true)
+-----+---+-----+
| name|age|score|
+-----+---+-----+
|Arike| 28| 78.6|
|  Bob| 32|45.32|
|Corry| 65|98.47|
+-----+---+-----+

createDataFrame method

A different approach to create DataFrame is to utilize createDataFrame API. As before we use parallelize to create a RDD but performing a map transformation and specifying the column types. 

list_of_persons = [('Arike', 28, 78.6), ('Bob', 32, 45.32), ('Corry', 65, 98.47)]
rdd = sc.parallelize(list_of_persons)
person = rdd.map(lambda x: Row(name=x[0], age=int(x[1]), score=float(x[2])))
schemaPeople = sqlContext.createDataFrame(person)
schemaPeople.printSchema()
schemaPeople.show()


root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)
 |-- score: double (nullable = true)
+---+-----+-----+
|age| name|score|
+---+-----+-----+
| 28|Arike| 78.6|
| 32|  Bob|45.32|
| 65|Corry|98.47|
+---+-----+-----+

read method

Reading a file from the disk will return a DataFrame. Spark supports multiple formats like CSV, JSON or parquet to name a few. For more information on how to read a file please see these posts:
How to read a JSON file,
How to read a CSV file 

df = spark.read.csv("data.csv")
df.printSchema()
df.show()
df = spark.read.json("data.json")
df.printSchema()
df.show()
df = spark.read.parquet("data_parquet")
df.printSchema()
df.show()


root
 |-- _c0: string (nullable = true)
 |-- _c1: string (nullable = true)
 |-- _c2: string (nullable = true)
+-----+---+-----+
|  _c0|_c1|  _c2|
+-----+---+-----+
| Name|Age|Score|
|Arike| 28| 78.6|
|  Bob| 32|45.32|
|Corry| 65|98.47|
+-----+---+-----+
root
 |-- Age: long (nullable = true)
 |-- Name: string (nullable = true)
 |-- Score: double (nullable = true)
+---+-----+-----+
|Age| Name|Score|
+---+-----+-----+
| 28|Arike| 78.6|
| 32|  Bob|45.32|
| 65|Corry|98.47|
+---+-----+-----+
root
 |-- Age: long (nullable = true)
 |-- Name: string (nullable = true)
 |-- Score: double (nullable = true)
+---+-----+-----+
|Age| Name|Score|
+---+-----+-----+
| 28|Arike| 78.6|
| 32|  Bob|45.32|
| 65|Corry|98.47|
+---+-----+-----+

sql method

You can also use the SQL method to write multiple select SQL queries and concatenate them using union. 

spark.sql("(select 'Arike' as name, 28 as age, 78.6 as score) union " +
          "(select 'Bob' as name, 32 as age, 45.32 as score) union " +
          "(select 'Corry' as name, 65 as age, 98.47 as score)")
df.printSchema()
df.show()


root
 |-- Age: long (nullable = true)
 |-- Name: string (nullable = true)
 |-- Score: double (nullable = true)
+---+-----+-----+
|Age| Name|Score|
+---+-----+-----+
| 28|Arike| 78.6|
| 32|  Bob|45.32|
| 65|Corry|98.47|
+---+-----+-----+

Conclusion

I believe the practical approach is to use the read method to create a DataFrame. The other ways can be useful for small reference table you might join with.

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...
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

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