PySpark: Dataframe To DB

Pradeep

PySpark: Dataframe To DB

This tutorial will explain how to write data from Spark dataframe into various types of databases(such as Mysql, SingleStore, Teradata) using JDBC Connection.

DataFrameWriter's "write" function can be used to export data from Spark dataframe to database table.
Both option() and mode() functions can be used to alter the behavior of write operation but in different sense.
Visit write modes page to understand how mode function can be used to alter write behaviour when data/table already exists.
you can also visit dataframe options page to understand how option/options function can be used to alter other write behaviour.

Following topics will be covered in this page:

➠ Write To Mysql: This section will explain (with examples) how to write dataframe into Mysql database using JDBC connection. "mysql-connector-java-8.0.11.jar" jar should be present in Spark library to write data to Mysql database using JDBC connection. This jar can be downloaded from Mysql website.

Replace below attributes in the examples:

Host/port in the URL from Mysql.dbmstutorials.com:3306 to your server.
Username in the user from tutorial_user to your username.
Password value in the password from user_password to your password.
Tablename in the dbtable from tutorial_db.department to your table name.

Create new table: Below commands will throw exception if table is already present.

Example 1: Create new table using options() function.


df.write.format("jdbc").options(driver="com.mysql.cj.jdbc.Driver",
                                   user="tutorial_user",
                                   password="user_password",
                                   url="jdbc:mysql://Mysql.dbmstutorials.com:3306?serverTimezone=UTC&useSSL=false",
                                   dbtable="tutorial_db.spark_department"
                                   ).save()

Example 2: Create new table using option() function.


df.write.format("jdbc") \
        .option("driver","com.mysql.cj.jdbc.Driver") \
        .option("user","tutorial_user") \
        .option("password","user_password") \
        .option("url","jdbc:mysql://Mysql.dbmstutorials.com:3306?serverTimezone=UTC&useSSL=false") \
        .option("dbtable","tutorial_db.spark_department") \
        .save()

Append data to existing table:

Example 1: Add rows to existing table using append mode and options() function.


df.write.mode("append").format("jdbc").options(driver="com.mysql.cj.jdbc.Driver",
                                   user="tutorial_user",
                                   password="user_password",
                                   url="jdbc:mysql://Mysql.dbmstutorials.com:3306?serverTimezone=UTC&useSSL=false",
                                   dbtable="tutorial_db.spark_department"
                                   ).save()

Example 2: Add rows to existing table using append mode and option() function.


df.write.mode("append").format("jdbc") \
        .option("driver","com.mysql.cj.jdbc.Driver") \
        .option("user","tutorial_user") \
        .option("password","user_password") \
        .option("url","jdbc:mysql://Mysql.dbmstutorials.com:3306?serverTimezone=UTC&useSSL=false") \
        .option("dbtable","tutorial_db.spark_department") \
        .save()

Overwrite existing table:

Example 1: Overwrite existing table using overwrite mode and options() function.


df.write.mode("overwrite").format("jdbc").options(driver="com.mysql.cj.jdbc.Driver",
                                   user="tutorial_user",
                                   password="user_password",
                                   url="jdbc:mysql://Mysql.dbmstutorials.com:3306?serverTimezone=UTC&useSSL=false",
                                   dbtable="tutorial_db.spark_department"
                                   ).save()

Example 2: Overwrite existing table using overwrite mode and option() function.


df.write.mode("overwrite").format("jdbc") \
        .option("driver","com.mysql.cj.jdbc.Driver") \
        .option("user","tutorial_user") \
        .option("password","user_password") \
        .option("url","jdbc:mysql://Mysql.dbmstutorials.com:3306?serverTimezone=UTC&useSSL=false") \
        .option("dbtable","tutorial_db.spark_department") \
        .save()

Ignore Write Operation if table exists:

Example 1: Write operation can be ignored if table exists using ignore mode and option() function.


df.write.mode("ignore").format("jdbc") \
        .option("driver","com.mysql.cj.jdbc.Driver") \
        .option("user","tutorial_user") \
        .option("password","user_password") \
        .option("url","jdbc:mysql://Mysql.dbmstutorials.com:3306?serverTimezone=UTC&useSSL=false") \
        .option("dbtable","tutorial_db.spark_department") \
        .save()

➠ Write To SingleStore: This section will explain (with examples) how to write dataframe into SingleStore / MemSQL database using JDBC connection.

Replace below attributes in the examples:

Host/port in the URL from singlestore.dbmstutorials.com:3306 to your server.
Username in the user from tutorial_user to your username.
Password value in the password from user_password to your password.
Database in the URL from tutorial_db database to your database name.
Tablename in the dbtable from tutorial_db.department to your table name.

Create new table: Below command will throw exception if table is already present.


df.write.format("jdbc") \
        .option("driver","org.mariadb.jdbc.Driver") \
        .option("user","tutorial_user") \
        .option("password","user_password") \
        .option("url","jdbc:mysql://singlestore.dbmstutorials.com:3306/tutorial_db") \
        .option("dbtable","spark_department") \
        .save()

Append data to existing table: Add rows to existing table using append mode and option() function.


df.write.mode("append").format("jdbc") \
        .option("driver","org.mariadb.jdbc.Driver") \
        .option("user","tutorial_user") \
        .option("password","user_password") \
        .option("url","jdbc:mysql://singlestore.dbmstutorials.com:3306/tutorial_db") \
        .option("dbtable","spark_department") \
        .save()

Overwrite existing table: Overwrite existing table using overwrite mode and option() function.


df.write.mode("overwrite").format("jdbc") \
        .option("driver","org.mariadb.jdbc.Driver") \
        .option("user","tutorial_user") \
        .option("password","user_password") \
        .option("url","jdbc:mysql://singlestore.dbmstutorials.com:3306/tutorial_db") \
        .option("dbtable","tutorial_db.spark_department") \
        .save()

Ignore Write Operation if table exists: Write operation can be ignored if table exists using ignore mode and option() function.


df.write.mode("ignore").format("jdbc") \
        .option("driver","org.mariadb.jdbc.Driver") \
        .option("user","tutorial_user") \
        .option("password","user_password") \
        .option("url","jdbc:mysql://singlestore.dbmstutorials.com:3306/tutorial_db") \
        .option("dbtable","tutorial_db.spark_department") \
        .save()

➠ Write To Teradata: This section will explain (with examples) how to write dataframe into Teradata database using JDBC connection. "terajdbc4.jar" jar should be present in Spark library to write data from Teradata using JDBC connection. This jar can be downloaded from Teradata website.

Replace below attributes in the examples:

Host in the URL from Teradata.dbmstutorials.com to your server.
Username in the user from tutorial_user to your username.
Password value in the password from user_password to your password.
Tablename in the dbtable from tutorial_db.department to your table name.

Create new table: Below command will throw exception if table is already present.


df.write.format("jdbc") \
        .option("driver", "com.teradata.jdbc.TeraDriver") \
        .option("user", "tutorial_user") \
        .option("password", "user_password") \
        .option("url", "jdbc:teradata://Teradata.dbmstutorials.com/COLUMN_NAME=ON") \
        .option("dbtable", "tutorial_db.spark_department") \
        .save()

Append data to existing table: Add rows to existing table using append mode and option() function.


df.write.mode("append").format("jdbc") \
        .option("driver", "com.teradata.jdbc.TeraDriver") \
        .option("user", "tutorial_user") \
        .option("password", "user_password") \
        .option("url", "jdbc:teradata://Teradata.dbmstutorials.com/COLUMN_NAME=ON") \
        .option("dbtable", "tutorial_db.spark_department") \
        .save()

Overwrite existing table: Overwrite existing table using overwrite mode and option() function.


df.write.mode("overwrite").format("jdbc") \
        .option("driver", "com.teradata.jdbc.TeraDriver") \
        .option("user", "tutorial_user") \
        .option("password", "user_password") \
        .option("url", "jdbc:teradata://Teradata.dbmstutorials.com/COLUMN_NAME=ON") \
        .option("dbtable", "tutorial_db.spark_department") \
        .save()

Ignore Write Operation if table exists: Write operation can be ignored if table exists using ignore mode and option() function.


df.write.mode("ignore").format("jdbc") \
        .option("driver", "com.teradata.jdbc.TeraDriver") \
        .option("user", "tutorial_user") \
        .option("password", "user_password") \
        .option("url", "jdbc:teradata://Teradata.dbmstutorials.com/COLUMN_NAME=ON") \
        .option("dbtable", "tutorial_db.spark_department") \
        .save()

➠ Write To Hive: This section will explain (with examples) how to write data into hive table. If hive setup is present for Spark then saveAsTable() function can be used to write data into Hive table.

Create new table: Below command will throw exception if table is already present.
```
df.write.saveAsTable("retail.spark_department")
```

Append data to existing table: Add rows to existing table using append mode.


Example 1:
df.write.insertInto("retail.spark_department")


Example 2:
df.write.mode("append").saveAsTable("retail.spark_department")

Overwrite existing table: Overwrite existing table using overwrite mode and option() function.
```
df.write.mode("overwrite").saveAsTable("retail.spark_department")
```
Ignore Write Operation if table exists: Write operation can be ignored if table exists using ignore mode.
```
df.write.mode("ignore").saveAsTable("retail.spark_department")
```