Introduction
In PySpark, the createDataFrame() and toDF() methods are used to create a DataFrame manually. Using these methods, you can create a Spark DataFrame from a list, PySpark RDD, another DataFrame, and various other data sources.
You can also create a DataFrame from different sources like Text, CSV, JSON, XML, Parquet, Avro, ORC, Binary files, RDBMS Tables, Hive, HBase, and many more.
One easy way to create a Spark DataFrame manually is from an existing RDD. First, create an RDD from a collection by calling parallelize().
# Import
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.master("local[1]") \
.appName("AIStudyLabs.com") \
.getOrCreate()
# Create a list of tuples with some data
data = [("Java", "20000"), ("Python", "100000"), ("Scala", "3000")]
# Create Spark RDD from data
rdd = spark.sparkContext.parallelize(data)
Create DataFrame from RDD using toDF()
Once we have an RDD, let’s use toDF() to create a DataFrame. By default, it creates column names as “_1” and “_2” as we have two columns for each row.
# Create DataFrame from RDD
dfFromRDD1 = rdd.toDF()
# Print DataFrame schema
dfFromRDD1.printSchema()
# Output:
# root
# |-- _1: string (nullable = true)
# |-- _2: string (nullable = true)
# Show DataFrame
dfFromRDD1.show()
# Output:
# +------+------+
# | _1| _2|
# +------+------+
# | Java| 20000|
# |Python|100000|
# | Scala| 3000|
# +------+------+
Since RDD is schema-less without column names and data types, converting from RDD to a DataFrame gives you default column names as _1, _2 and so on, with data types as String.
DataFrame.printSchema() is used to print schema to the console. DataFrame.show() is used to print the contents of the DataFrame to the console.
Assign Column Names to DataFrame
RDD.toDF() has another signature to assign column names; this takes a variable number of arguments for column names, as shown below.
# Assign column names to DataFrame
dfFromRDD2 = rdd.toDF(["language", "users_count"])
dfFromRDD2.printSchema()
# root
# |-- language: string (nullable = true)
# |-- users_count: string (nullable = true)
dfFromRDD2.show()
# +--------+-----------+
# |language|users_count|
# +--------+-----------+
# | Java| 20000|
# | Python| 100000|
# | Scala| 3000|
# +--------+-----------+
Remember, here, we just assigned column names but not the data types. So, it still assigns all data types as Strings.
We can change this behavior by supplying a schema; using the schema, we can specify the column name, data type, and nullable status for each column/field. This will be explained in future sections.
Using createDataFrame() from SparkSession
Using createDataFrame() from SparkSession is another way to create a DataFrame. This method takes the rdd object as an argument and chains it with toDF() to specify the names of the columns.
# Using createDataFrame()
columns = ["language", "users_count"]
dfFromRDD3 = spark.createDataFrame(rdd).toDF(*columns)
Create DataFrame from List and Collection
Let's see some examples of creating a Spark DataFrame directly from a list without using RDD. These examples are similar to what we have seen above with RDD, but we use the data object instead of the rdd object.
Using toDF() on List
toDF() on a list creates a DataFrame. Ensure to use this within a session context that includes implicits for DataFrame transformations.
data = [("Java", "20000"), ("Python", "100000"), ("Scala", "3000")]
dfFromData4 = spark.createDataFrame(data).toDF("language", "users_count")
Using createDataFrame() from SparkSession
Calling createDataFrame() from SparkSession is another way to create a DataFrame. It takes a collection object as an argument and chains it with toDF() to specify the column names.
# From Data (USING createDataFrame)
dfFromData2 = spark.createDataFrame(data, columns)
Create Spark DataFrame from Files
In all the above examples, you have learned how to create a Spark DataFrame from an RDD and data collection objects. In real-time, these are less used. In this section, you will learn how to create a DataFrame from data sources like CSV, text, JSON, Avro, etc.
Create Spark DataFrame from CSV
Spark, by default, provides an API to read delimiter files like comma, pipe, and tab-separated files. It also provides several options for handling files with headers, without headers, double quotes, data types, etc.
# Create Spark DataFrame from CSV
df2 = spark.read.csv("/src/resources/file.csv")
In later chapters, more details will be provided on reading CSV files with various options.
Creating from text (TXT) file
To create a DataFrame from text files.
# Creating from text (TXT) file
df2 = spark.read.text("/src/resources/file.txt")
Creating from JSON file
To create a DataFrame from JSON files.
# Create from JSON file
df2 = spark.read.json("/src/resources/file.json")
Creating from MySQL table
Ensure you have the necessary MySQL connector library in your environment.
# From MySQL table
df_mysql = spark.read.format("jdbc") \
.option("url", "jdbc:mysql://localhost:port/db") \
.option("driver", "com.mysql.jdbc.Driver") \
.option("dbtable", "tablename") \
.option("user", "user") \
.option("password", "password") \
.load()
In later chapters, there will be further explanations on reading CSV files with several options.
Creating from Hive table
The following is an example of reading a Hive table. More details will be provided in later chapters.
# Change default warehouse location and metastore
spark = SparkSession.builder.master("local[*]") \
.appName("SparkCreateTableExample") \
.config("spark.sql.warehouse.dir", "/hive/warehouse/dir") \
.config("hive.metastore.uris", "thrift://remote-host:9083") \
.enableHiveSupport() \
.getOrCreate()
# Read Hive table
df = spark.sql("select * from employee")
df.show()
Besides these, you can also create a DataFrame from many other sources like AWS S3, Avro, Parquet, ORC, etc.