Introduction
In Spark, the createDataFrame() and toDF() methods are used to create a DataFrame manually. Using these methods, you can create a Spark DataFrame from a List, Seq data object, RDD, DataFrame, and Dataset.
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 Spark DataFrame manually is from an existing RDD. First, create an RDD from a collection Seq by calling parallelize().
// Import
import org.apache.spark.sql.SparkSession
val spark: SparkSession = SparkSession.builder()
.master("local[1]")
.appName("AIStudyLabs.com")
.getOrCreate()
// Create Seq object with some data
val data = Seq(("Java", "20000"), ("Python", "100000"), ("Scala", "3000"))
// Create Spark RDD from data
val 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
import spark.implicits._
val 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 type, converting from RDD to DataFrame gives you default column names as _1, _2 and so on and data type 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.
Note: In the above example, import spark.implicits._: This is a wildcard import that brings all implicit methods and conversions defined in the implicits package into scope. These implicits include methods for converting standard Scala types to Spark SQL types, enabling DataFrame operations, and providing syntactic sugar for DataFrame transformations and actions.
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
val 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 schema; Using the schema, we can specify the column name, data type, and nullable for each column/field. I will explain this 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 chain it with toDF() to specify the names of the columns.
// Using CreateDataFrame()
val columns = Seq("language", "users_count")
val dfFromRDD3 = spark.createDataFrame(rdd).toDF(columns: _*)
Here, columns:_* is a shorthand syntax in Scala used to pass an array of values as arguments. It takes the values stored in the columns array (which presumably contains the column names) and passes them to the toDF method.
Create DataFrame from List and Seq Collection
Let's see some examples of creating Spark DataFrame directly from the collection Seq[T] or List[T] without using RDD. These examples would be similar to what we have seen in the above section with RDD, but we use “data” object instead of “rdd” object.
Using toDF() on List or Seq collection
toDF() on collection (Seq, List) object creates a DataFrame. Make sure to import import spark.implicits._ to use toDF().
import spark.implicits._
val data = Seq(("Java", "20000"), ("Python", "100000"), ("Scala", "3000"))
val dfFromData4 = data.toDF()
Using createDataFrame() from SparkSession
Calling createDataFrame() from SparkSession is another way to create. It takes a collection object (Seq or List) as an argument and chain it with toDF() to specify the column names.
// From Data (USING createDataFrame)
var dfFromData2 = spark.createDataFrame(data).toDF(columns: _*)
Create Spark DataFrame from Files
In all the above examples, you have learned Spark to create DataFrame from RDD and data collection objects. In real-time, these are less used, In this section, you will learn how to create DataFrame from data sources like CSV, text, JSON, Avro e.t.c
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
val df2 = spark.read.csv("/src/resources/file.csv")
In later chapters, I will explain in more detail how to read CSV files with several options.
Creating from text (TXT) file
To create a DataFrame from text files.
// Creating from text (TXT) file
val df2 = spark.read
.text("/src/resources/file.txt")
Creating from JSON file
To create a DataFrame from JSON files.
// Create from JSON file
val df2 = spark.read
.json("/src/resources/file.json")
Creating from Mysql table
Make sure you have MySQL library as a dependency in your pom.xml file or MySQL jars in your classpath.
// From Mysql table
val 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, I will explain in more detail how to read CSV files with several options.
Creating from Hive table
The following is an example of reading a Hive table. I will explain this in detail in later chapters.
// Change default warehouse location and metastore
val 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
val df = spark.sql("select * from employee")
df.show()
Besides these, you can also create a DataFrame from many other sources like AWS S3, Reading from Avro, Parquet, ORC e.t.c