RxJava is an awesome tool for reactive programming that’s also useful as a Java CSV reader.
In case you’ve never used or even heard of RxJava, it is a a way of programming with asynchronous data streams. It is also event-based programming where you can observe or listen for stream data and do something when said stream is discovered. Data streams can be anything ranging from variables to data structures.
For this example, lets say a line of CSV data is a data stream. Many developers find themselves needing to load CSV data into their database more often than they expect. In most scenarios you cannot just read lines from a CSV file and drop it into a database. In this example, we’re going to use RxJava to subscribe to a stream of data and transform it into Java for reading the CSV files that will be saved into Couchbase.
Os requisitos
There are not too many requirements to get this project up and running. At a minimum, you’ll need the following:
- JDK 1.8+
- Apache Maven 3.3+
- Servidor Couchbase 4.1+
All of the development and work will happen with the JDK 1.8 and Maven, and this includes running the application.
Entendendo o conjunto de dados e o modelo de dados
A great way to get your feet wet when it comes to RxJava is to get a sample dataset to play around with. For simplicity we’ll invent our own comma separated value (CSV) file, but if you wanted something more extravagant, you can visit the data science website, Kaggle.
Let’s assume that our simple CSV dataset has the following columns per row:
- id
- primeiro_nome
- sobrenome
From a query and analysis perspective, working with the data in CSV format is near impossible. Instead, this data is going to be stored as NoSQL data so it can be later processed. We won’t get into the number crunching and querying here, but it will come in a future article. Right now we just want to get it into NoSQL format.
Quando carregado no Couchbase, cada linha do CSV terá uma aparência semelhante à seguinte:
|
1 2 3 4 5 6 7 8 |
{ "id": 1, "first_name": "Nic", "last_name": "Raboy", "twitter": "nraboy" } |
Yes, the above chunk of data is a JSON document, which is what Couchbase supports. Now that we know the data goals, we can begin loading the CSV data into Couchbase with RxJava.
Transformando os dados brutos e gravando no Couchbase
To use RxJava to load CSV data via a Java application, a few dependencies must be included. We need to include RxJava, OpenCSV, and the Couchbase Java SDK. Since we’re using Maven, all can be included via the Maven pom.xml file. To include RxJava, include the following dependency in your Maven file:
|
1 2 3 4 5 6 7 |
io.reativox rxjava 1.1.2 |
Since the raw data will be in the form of CSV, we can use the open source CSV library for Java called OpenCSV. The Maven dependency for OpenCSV can be added like this:
|
1 2 3 4 5 6 7 |
com.opencsv opencsv 3.7 |
Finally, Java needs to be connected to Servidor Couchbase. This can be done through the SDK Java do Couchbase. To add this dependency into your Maven project, add the following to your pom.xml file:
|
1 2 3 4 5 6 7 |
com.couchbase.cliente java-cliente 2.2.0 |
Todas as dependências do projeto estão prontas para funcionar!
To load, but not read, the CSV file, you’ll create a new CSVReader object as follows:
|
1 2 3 |
CSVReader leitor = novo CSVReader(novo FileReader("PATH_TO_CSV_FILE")); |
Since this data will eventually be processed into Couchbase, we must connect to our server and open our bucket.
|
1 2 3 |
Balde balde = CouchbaseCluster.criar("http://localhost:8091").openBucket("default", ""); |
The above assumes that Couchbase is running locally and the data will be saved in the default bucket without a password.
To process the CSV dataset, an RxJava Observable can be created:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
Observável .de(leitor) .mapa( csvRow -> { JsonObject objeto = JsonObject.criar(); objeto .colocar("first_name", csvRow[1]) .colocar("last_name", csvRow[2]) .colocar("twitter", csvRow[3]); retorno JsonDocument.criar(csvRow[0], objeto); } ) .assinar(documento -> balde.upsert(documento), erro -> Sistema.fora.println(erro)); |
To break down what is happening in the Observable, the following steps occur.
O CSVReader creates an Iterable<String[]>. The Observable will use the Iterable<String[]> as the source of data using the .from método.
The data read will be an array of strings, not something that can be stored directly in the database. Using the .mapa function, the array of strings can be transformed into whatever we decide. In this case, the goal is to map each line of the CSV to a Couchbase document. During this mapping process we can do further data cleanup. For example, we could do something like csvRow[*].trim() to strip out any leading and trailing whitespace in each CSV column.
Finally with each read line processed, it must be saved to Couchbase. The .subscribe method will subscribe to notifications that the Observable emits, in this case the manipulated data.
Conclusão
You just got a taste of loading dirty CSV data into Couchbase by using RxJava and the Couchbase Java SDK. By doing reactive programming you can take actions on any data stream that you’re observing. In our scenario we just wanted to load a CSV file into Couchbase.
If you’ve got a massive dataset, you can take this tutorial to the next level by using Apache Spark. I wrote a very similar CSV loader found aqui that makes use of Spark.
