I recently saw a question on our forums where someone wanted to move some data from CouchDB to Couchbase. Since I usually help friends in need when they are moving, I thought I would help. Moving requires preparation, especially with large furniture like Couch :D<\/p>\n
Sorry about that metaphor, now let me explain how I did it. My goal is to move data from CouchDB to Couchbase. So the first question is how to get the data out of CouchDB? There are several options available here and the most straightforward for me was to use the REST API. If you use the _all_docs endpoint with the include_doc parameter set to true, you will get every document. This is exactly what I need.<\/p>\n
Now instead of downloading the REST response directly, I can use the Java 8 stream API. And since I will be using RxJava as it\u2019s part of our SDK, I need to wrap that stream in an observable. It\u2019s actually quite simple:<\/p>\n
\u00a0<\/p>\n
\n\nURL url = new URL(couchDBRequest);\nHttpURLConnection conn = (HttpURLConnection) url.openConnection();\nconn.setRequestMethod(\u201cGET\u201d);\nconn.setRequestProperty(\u201cAccept\u201d, \u201capplication\/json\u201d);\n\/\/assume this is going to be a big file\u2026\nconn.setReadTimeout(0);\nif (conn.getResponseCode() != 200) {\n throw new RuntimeException(\u201cFailed : HTTP error code : \u201d + conn.getResponseCode());\n}\nBufferedReader inp2 = new BufferedReader(new InputStreamReader(conn.getInputStream()));\nfinal long[] totalRows = new long[2];\nint count = Observable.from(inp2.lines()::iterator).count().toBlocking().single(); \n<\/code><\/pre>\n\u00a0<\/p>\n
\u00a0<\/p>\n
With this I have an observable that emits CouchDB documents. Next step is of course to take these documents and send them to Couchbase. I will be using a flatMap first to parse each line of the response. It\u2019s easy to do because each line of the response contains one doc as you can see:<\/p>\n
\u00a0<\/p>\n
\n\n{\"total_rows\":2,\"offset\":0,\"rows\":[\n {\"id\":\"f750a4273b48b6c1146fe4ead1000c1b\",\"key\":\"f750a4273b48b6c1146fe4ead1000c1b\",\"value\":{\"rev\":\"2-bea53b374bf5a427ab15245bc029cac0\"},\"doc\":{\"_id\":\"f750a4273b48b6c1146fe4ead1000c1b\",\"_rev\":\"2-bea53b374bf5a427ab15245bc029cac0\",\"title\":\"A title\"}},\n {\"id\":\"f750a4273b48b6c1146fe4ead1000c24\",\"key\":\"f750a4273b48b6c1146fe4ead1000c24\",\"value\":{\"rev\":\"1-9baa68f46c29940ad7a6d57ae1a04002\"},\"doc\":{\"_id\":\"f750a4273b48b6c1146fe4ead1000c24\",\"_rev\":\"1-9baa68f46c29940ad7a6d57ae1a04002\",\"title\":\"Another Title\"}}\n]}\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 \n<\/code><\/pre>\nI have to handle the first and last line separately. In the first line I get the total_rows and the offset information. I don\u2019t have to do anything at the last line. I simply return Observable.empty() for those two lines as I don\u2019t have anything to feed to the next operator. All the other lines contain a row edited by CouchDB. Each of these rows contains a JSON document we can wrap in a JsonNode.<\/p>\n
The next operator is also a flatMap. Here I extract the key of the document and its content as a String. Since I have a Json object as a String, I don\u2019t need any kind of mapping using Jackson or the like. I can directly use a RawJsonDocument. Once I have the RawJsonDocument I can import it into Couchbase. To do so I use the upsert method. It\u2019s a bit like a \u201cno questions asked\u201d thing. We don\u2019t care if the document exists or not: if it doesn\u2019t exist then it will be created, whereas if it does exist then it will be replaced. This might not be the behaviour you want but it\u2019s the simplest for this scenario as it means I don\u2019t have to handle errors when a key already exists.<\/p>\n
I assign a 500 millisecond timeout to the operation as it really should not take long. Then I use the RetryBuilder. It\u2019s a nice helper added by Simon Basle to easily manage retry on error. Here I retry up to 100 times if I get a RequestCancelledException. I add an arbitrary delay of 31 seconds before each retry. I do the same for the TemporaryFailureException and BackpressureException. Here I use a delay of a 100 milliseconds instead.<\/p>\n
Then I use doOnError and doOnNext to log the key of the document in either the success log file or the error log file. doOnX methods don\u2019t change the core semantics of your stream (like transforming data does, for instance) but rather add some behavior on the side, \u201cside effects\u201d. In there I write a String to a log file using FileWriter. This is unfortunately synchronous and blocking. I might change this to use an async logger instead.<\/p>\n
I then use onErrorResumeNext to make sure the import continues even if there are errors. Finally I use count().toBlocking().single() to know how many insert I did in Couchbase. I compare the result to the number of total_rows at the end.<\/p>\n
In the end the code looks like this:<\/p>\n