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 a Spark DataFrame manually is from an existing RDD. First, create an RDD from a collection using JavaSparkContext.parallelize().

// Import
import org.apache.spark.sql.SparkSession;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.JavaRDD;
import java.util.Arrays;
import java.util.List;

SparkSession spark = SparkSession.builder()
        .master("local[1]")
        .appName("AIStudyLabs.com")
        .getOrCreate();

// Create List object with some data
List<Tuple2<String, String>> data = Arrays.asList(
        new Tuple2<>("Java", "20000"),
        new Tuple2<>("Python", "100000"),
        new Tuple2<>("Scala", "3000"));

// Create JavaRDD from data
JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
JavaRDD<Tuple2<String, String>> rdd = jsc.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.

import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.Encoders;

// Create DataFrame from RDD
Dataset<Row> dfFromRDD1 = spark.createDataset(rdd.rdd(), Encoders.tuple(Encoders.STRING(), Encoders.STRING()))
        .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.

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.

import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

// Define the schema
StructType schema = new StructType(new StructField[]{
    DataTypes.createStructField("language", DataTypes.StringType, true),
    DataTypes.createStructField("users_count", DataTypes.StringType, true)
});

// Assign column names to DataFrame
Dataset<Row> dfFromRDD2 = spark.createDataFrame(rdd, schema);

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 you can chain it with withColumnRenamed() to specify the names of the columns.

// Create DataFrame using createDataFrame()
Dataset<Row> dfFromRDD3 = spark.createDataFrame(rdd, schema);

Create DataFrame from List

Let's see some examples of creating Spark DataFrame directly from the List 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

toDF() on a List object creates a DataFrame.

// Create DataFrame from List
Dataset<Row> dfFromData4 = spark.createDataFrame(data, schema);

Create Spark DataFrame from Files

In all the above examples, you have learned how Spark creates DataFrame from RDD and data collection objects. In real-time, these are less used. In this section, you will learn how to create DataFrames 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
Dataset<Row> 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
Dataset<Row> df2 = spark.read()
        .text("/src/resources/file.txt");

Creating from JSON file

To create a DataFrame from JSON files.

// Create from JSON file
Dataset<Row> 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
Dataset<Row> df_mysql = spark.read().format("jdbc")
        .option("url", "jdbc:mysql://localhost:port/db")
        .option("driver", "com.mysql.cj.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
SparkSession 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
Dataset<Row> 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, etc.