Upsert Records to Oracle

This example will show you how you can use DataPipline to upsert records to Oracle.

To accomplish this both JdbcUpsertWriter and OracleUpsert will be used.

This example can be easily modified to Upsert Records to MySql or MariaDB or Upsert Records to PostgreSql.

Input CSV file

Account,LastName,FirstName,Balance,CreditLimit,AccountCreated,Rating
101,Reeves,Keanu,9315.45,10000.00,1998-1-17,A
312,Butler,Gerard,90.00,1000.00,2003-8-6,B
101,Hewitt,Jennifer Love,0,17000.00,1985-5-25,B
312,Pinkett-Smith,Jada,49654.87,100000.00,2006-12-5,A
317,Murray,Bill,789.65,5000.00,2007-2-5,C
317,Murray,Bill,1,5000.00,2007-2-5,A

Java code listing

package com.northconcepts.datapipeline.examples.cookbook;

import java.io.File;
import java.io.OutputStreamWriter;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

import com.northconcepts.datapipeline.core.DataReader;
import com.northconcepts.datapipeline.core.DataWriter;
import com.northconcepts.datapipeline.csv.CSVReader;
import com.northconcepts.datapipeline.csv.CSVWriter;
import com.northconcepts.datapipeline.jdbc.JdbcReader;
import com.northconcepts.datapipeline.jdbc.JdbcUpsertWriter;
import com.northconcepts.datapipeline.jdbc.upsert.OracleUpsert;
import com.northconcepts.datapipeline.job.Job;
import com.northconcepts.datapipeline.transform.BasicFieldTransformer;
import com.northconcepts.datapipeline.transform.TransformingReader;

public class UpsertRecordsToOracle {

	public static void main(String[] args) throws Throwable {

		final String DATABASE_DRIVER = "oracle.jdbc.OracleDriver";
		final String DATABASE_URL = "jdbc:oracle:thin:system/oracle@localhost:1521:xe";
		final String DATABASE_USERNAME = "system";
		final String DATABASE_PASSWORD = "oracle";
		final String DATABASE_TABLE = "CreditBalance";
		final String PRIMARY_KEY = "Account";

		Class.forName(DATABASE_DRIVER);
		Connection connection = DriverManager.getConnection(DATABASE_URL, DATABASE_USERNAME, DATABASE_PASSWORD);

		createTable(connection);

		DataReader reader = new CSVReader(new File("example/data/input/credit-balance-02.csv")).setFieldNamesInFirstRow(true);
		reader = new TransformingReader(reader).add(
				new BasicFieldTransformer("Account").stringToInt(),
				new BasicFieldTransformer("FirstName"),
				new BasicFieldTransformer("LastName"),
				new BasicFieldTransformer("Balance").stringToFloat(),
				new BasicFieldTransformer("CreditLimit").stringToFloat(),
				new BasicFieldTransformer("AccountCreated").stringToDate("yyyy-MM-dd"),
				new BasicFieldTransformer("Rating").stringToChar());

		DataWriter writer = new JdbcUpsertWriter(connection, DATABASE_TABLE, new OracleUpsert(), PRIMARY_KEY);

		Job.run(reader, writer);

		reader = new JdbcReader(connection, "Select * from CreditBalance");
		writer = new CSVWriter(new OutputStreamWriter(System.out));

		Job.run(reader, writer);
	}

	public static void createTable(Connection connection) throws Throwable{
		PreparedStatement preparedStatement;
		String createTableQuery = "CREATE TABLE CreditBalance ("
			+ "Account NUMBER(10),"
			+ "LastName VARCHAR(256),"
			+ "FirstName VARCHAR(256),"
			+ "Balance NUMBER(19,4),"
			+ "CreditLimit NUMBER(19,4),"
			+ "AccountCreated DATE,"
			+ "Rating CHAR,"
			+ "PRIMARY KEY (Account)"
			+ ")";

		preparedStatement = connection.prepareStatement(createTableQuery);
		preparedStatement.execute();
	}
}

Code Walkthrough

1. First you need to feed your database details.
2. DriverManager is used to create a connection to the database.
3. createTable(connection) creates a table CreditBalance .
4. A CSVReader is created using the file path of the input file credit-balance-02.csv.
5. The CSVReader.setFieldNamesInFirstRow(true) method is invoked to specify that the names specified in the first row should be used as field names.
6. TransformingReader is used to apply transformation to the records passing through. e.g. In this case it is used to transform Account from string to Int.
7. JdbcUpsertWriter will write the records to the database table. For this example JdbcUpsertWriter takes four arguments i.e. Connection(connection), Table Name(CreditBalance), strategy(OracleUpsert()) and keyFieldNames(Account).
8. Data is transferred from the reader to writer via Job.run() method.
9. JdbcReader is used to obtain the records from the database using the query Select * from CreditBalance.
10. CSVWriter is used to write the CSV record to the output console
11. Job.run(reader, writer) is used to transfer the data from the reader to the writer.

JdbcUpsertWriter

Writes records to a database table using the upsert idiom (attempt to insert, but update if the duplicate key exists). The default strategy used is GenericUpsert, however, several others are provided: MergeUpsert, VariableFieldsUpsert, and MySqlUpsert.

CSVReader

CSVReader is an input reader which can be used to read CSV files. It is a sub-class of TextReader and inherits the open and close among other methods. The CSVReader.setFieldNamesInFirstRow(true) method causes the CSVReader to use the names specified in the first row of the input data as field names. If this method is not invoked, the fields would be named as A1, A2, etc. similar to MS Excel. If those fields names need to be changed, a rename transformation can be added on top of CSVReader or any other type (Refer Rename a field for example).

Output

The output will be written to both the console and the CreditBalance table.

Console Output

Account,LastName,FirstName,Balance,CreditLimit,AccountCreated,Rating
101,Reeves,Keanu,9315.45,10000.0,1998-01-17,A
312,Butler,Gerard,90.0,1000.0,2003-08-06,B
101,Hewitt,Jennifer Love,0.0,17000.0,1985-05-25,B
312,Pinkett-Smith,Jada,49654.87,100000.0,2006-12-05,A
317,Murray,Bill,789.65,5000.0,2007-02-05,C
317,Murray,Bill,1.0,5000.0,2007-02-05,A