Generating Unigram and Bigrams into MySQL from Hadoop SequenceFiles

In my previous post, I described how I used GNU Parallel to read a fairly large Lucene index into a set of Hadoop SequenceFiles. The objective is to use the data in the index to build a Unigram and Bigram Language Model for a spelling corrector. Since the spelling correction code is going to be called from a web application, I figured a good place to store the unigrams and bigrams in a MySQL database.

This is a fairly trivial task from the point of view of writing Map-Reduce code (the unigram writer is just a minor variation of the WordCount example), but this is the first time I was using Map-Reduce to crunch through a reasonably large dataset. I was also running Hadoop on a single large machine in pseudo-distributed mode, unlike previously where I mostly used it in local mode to build little proofs of concept. So there were certain things I learned about running Hadoop, which I will mention as they come up. But first, the code.

Java Code

As stated above, the code for both the UnigramCounter and BigramCounter are fairly trivial examples of Map-Reduce code. But I include them anyway, for completeness.

UnigramCounter.java

The UnigramCounter Mapper splits up the input text into words and writes them out to the context, where the Reducer picks them up and aggregates the counts, computes the soundex and metaphone values for the word, and writes the record out to a database table. The soundex and metaphones are for finding sound-alikes - I am not sure which one will give me the best results, so I compute both.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171

package com.mycompany.spell3.train; import java.io.IOException; import java.sql.Connection; import java.sql.DriverManager; import java.sql.PreparedStatement; import java.sql.SQLException; import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.codec.language.Metaphone; import org.apache.commons.codec.language.Soundex; import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; public class UnigramCounter extends Configured implements Tool { private static final String PROP_DBNAME = "dbname"; private static final String PROP_DBUSER = "dbuser"; private static final String PROP_DBPASS = "dbpass"; private static final String NULL_PATH = "/prod/hadoop/dummy"; public static class MapClass extends Mapper<LongWritable,Text,Text,IntWritable> { private static final IntWritable ONE = new IntWritable(1); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String s = StringUtils.lowerCase(value.toString()); String[] words = s.split("[^a-z]+"); for (String word : words) { context.write(new Text(word), ONE); } } } public static class ReduceClass extends Reducer<Text,IntWritable,Text,IntWritable> { private String MYSQL_DB_DRIVER = "com.mysql.jdbc.Driver"; private Connection conn; private PreparedStatement ps; private AtomicInteger counter = new AtomicInteger(0); private Soundex soundex; private Metaphone metaphone; @Override protected void setup(Context context) throws IOException, InterruptedException { try { Class.forName(MYSQL_DB_DRIVER); Configuration conf = context.getConfiguration(); conn = DriverManager.getConnection( "jdbc:mysql://localhost:3306/" + conf.get(PROP_DBNAME), conf.get(PROP_DBUSER), conf.get(PROP_DBPASS)); conn.setAutoCommit(false); ps = conn.prepareStatement( "insert into unigram_counts(word,soundex,metaphone,cnt) " + "values (?,?,?,?)"); soundex = new Soundex(); metaphone = new Metaphone(); } catch (Exception e) { throw new IOException(e); } } @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable value : values) { sum += value.get(); } insertToDb(key.toString(), sum); } private void insertToDb(String word, int count) throws IOException { try { ps.setString(1, word); ps.setString(2, soundex.soundex(word)); ps.setString(3, metaphone.metaphone(word)); ps.setInt(4, count); ps.execute(); int current = counter.incrementAndGet(); if (current % 1000 == 0) { conn.commit(); } } catch (SQLException e) { System.out.println("Failed to insert unigram: " + word); e.printStackTrace(); } } @Override protected void cleanup(Context context) throws IOException, InterruptedException { if (ps != null) { try { ps.close(); } catch (SQLException e1) {} } if (conn != null) { try { conn.commit(); conn.close(); } catch (SQLException e) { throw new IOException(e); } } } } @Override public int run(String[] args) throws Exception { Path input = new Path(args[0]); Path output = new Path(NULL_PATH); Configuration conf = getConf(); conf.set(PROP_DBNAME, args[1]); conf.set(PROP_DBUSER, args[2]); conf.set(PROP_DBPASS, args[3]); Job job = new Job(conf, "Unigram-Counter"); FileInputFormat.setInputPaths(job, input); FileOutputFormat.setOutputPath(job, output); job.setJarByClass(UnigramCounter.class); job.setMapperClass(MapClass.class); job.setReducerClass(ReduceClass.class); job.setInputFormatClass(SequenceFileInputFormat.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); job.setNumReduceTasks(5); boolean succ = job.waitForCompletion(true); if (! succ) { System.out.println("Job failed, exiting"); return -1; } return 0; } public static void main(String[] args) throws Exception { if (args.length != 4) { System.out.println( "Usage: UnigramCounter path_to_seqfiles output_db db_user db_pass"); System.exit(-1); } int res = ToolRunner.run(new Configuration(), new UnigramCounter(), args); System.exit(res); } }

BigramCounter.java

The BigramCounter Mapper uses a Sentence BreakIterator to break the input up into sentences, computes bigrams of word pairs within each sentence and writes them out to the context, where the Reducer picks them up, aggregates the counts and writes the bigram and count to another database table.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192

package com.mycompany.spell3.train; import java.io.IOException; import java.sql.Connection; import java.sql.DriverManager; import java.sql.PreparedStatement; import java.sql.SQLException; import java.text.BreakIterator; import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; public class BigramCounter extends Configured implements Tool { private static final String PROP_DBNAME = "dbname"; private static final String PROP_DBUSER = "dbuser"; private static final String PROP_DBPASS = "dbpass"; private static final String NULL_PATH = "/prod/hadoop/dummy"; public static class MapClass extends Mapper<LongWritable,Text,Text,IntWritable> { private static final IntWritable ONE = new IntWritable(1); private static final String SENTENCE_START = "<s>"; private static final String SENTENCE_END = "</s>"; private static final String WORD_SEPARATOR = "__"; @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String s = value.toString(); BreakIterator sit = BreakIterator.getSentenceInstance(); sit.setText(s); int start = sit.first(); int end = -1; while ((end = sit.next()) != BreakIterator.DONE) { String sentence = StringUtils.lowerCase(s.substring(start, end)); start = end; String[] words = sentence.split("[^a-z]+"); String prevWord = null; for (int i = 0; i < words.length; i++) { String bigram = null; if (i == 0) { // begin sentence bigram = StringUtils.join( new String[] {SENTENCE_START, words[i]}, WORD_SEPARATOR); } else if (i == words.length - 1) { // end sentence bigram = StringUtils.join( new String[] {words[i], SENTENCE_END}, WORD_SEPARATOR); } else { // middle of sentence bigram = StringUtils.join(new String[] { prevWord, words[i]}, WORD_SEPARATOR); } context.write(new Text(bigram), ONE); prevWord = words[i]; } } } } public static class ReduceClass extends Reducer<Text,IntWritable,Text,IntWritable> { private static final String MYSQL_DB_DRIVER = "com.mysql.jdbc.Driver"; private Connection conn; private PreparedStatement ps; private AtomicInteger counter = new AtomicInteger(0); @Override protected void setup(Context context) throws IOException, InterruptedException { try { Class.forName(MYSQL_DB_DRIVER); Configuration conf = context.getConfiguration(); conn = DriverManager.getConnection( "jdbc:mysql://localhost:3306/" + conf.get(PROP_DBNAME), conf.get(PROP_DBUSER), conf.get(PROP_DBPASS)); conn.setAutoCommit(false); ps = conn.prepareStatement( "insert into bigram_counts(bigram,cnt) values (?,?)"); } catch (Exception e) { throw new IOException(e); } } @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable value : values) { sum += value.get(); } insertToDb(key.toString(), sum); } private void insertToDb(String bigram, int sum) throws IOException { try { ps.setString(1, bigram); ps.setInt(2, sum); ps.execute(); int current = counter.incrementAndGet(); if (current % 1000 == 0) { conn.commit(); } } catch (SQLException e) { System.out.println("Failed to insert bigram: " + bigram); e.printStackTrace(); } } @Override protected void cleanup(Context context) throws IOException, InterruptedException { if (ps != null) { try { ps.close(); } catch (SQLException e) {} } if (conn != null) { try { conn.commit(); conn.close(); } catch (SQLException e) { throw new IOException(e); } } } } @Override public int run(String[] args) throws Exception { Path input = new Path(args[0]); Path output = new Path(NULL_PATH); Configuration conf = getConf(); conf.set(PROP_DBNAME, args[1]); conf.set(PROP_DBUSER, args[2]); conf.set(PROP_DBPASS, args[3]); Job job = new Job(conf, "Bigram-Counter"); FileInputFormat.setInputPaths(job, input); FileOutputFormat.setOutputPath(job, output); job.setJarByClass(BigramCounter.class); job.setMapperClass(MapClass.class); job.setReducerClass(ReduceClass.class); job.setInputFormatClass(SequenceFileInputFormat.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); job.setNumReduceTasks(5); boolean succ = job.waitForCompletion(true); if (! succ) { System.out.println("Job failed, exiting"); return -1; } return 0; } public static void main(String[] args) throws Exception { if (args.length != 4) { System.out.println( "Usage: BigramCounter path_to_seqfiles output_db db_user db_pass"); System.exit(-1); } int res = ToolRunner.run(new Configuration(), new BigramCounter(), args); System.exit(res); } }

Hadoop Configuration Changes

Hadoop is built to run on clusters of many medium size machines. What I had instead was one large 16-CPU machine, so I wanted to make sure that its processing power was utilized to the maximum possible. So I made the following changes to mapred-site.xml based on the advice in this StackOverflow page.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

<!-- Source: $HADOOP_HOME/conf/mapred-site.xml --> <configuration> ... <property> <name>mapred.tasktracker.map.tasks.maximum</name> <value>10</value> <description/> </property> <property> <name>mapred.tasktracker.reduce.tasks.maximum</name> <value>10</value> <description/> </property> </configuration>

In core-site.xml, I changed the location of the hadoop.tmp.dir to a large, relatively unused partition on the box instead of its default location. This was actually in response to a job failure where it ran out of HDFS space. Since at that point I had to rerun the job again anyway, I shut down Hadoop, deleted the old hadoop.tmp.dir and then restarted Hadoop and reformatted the namenode.

1 2 3 4 5 6 7 8 9 10

<!-- Source; $HADOOP_HOME/conf/core-site.xml --> <configuration> <property> <name>hadoop.tmp.dir</name> <value>/prod/hadoop/tmp</value> <description>A base for other temporary directories.</description> </property> ... </configuration>

Since I have only a single data node, I set the dfs.replication in hdfs-site.xml to 1.

1 2 3 4 5 6 7 8 9 10

<!-- $HADOOP_HOME/conf/hdfs-site.xml --> <configuration> <property> <name>dfs.replication</name> <value>1</value> <description/> </property> </configuration>

MySQL Configuration Changes

The default location of the MySQL data directory was in /var/lib/mysql, which was in the "/" partition, too small for my purposes. I actually ran out of disk space in this partition while writing bigrams to MySQL (the job just hangs at a fixed map-reduce completion status). I had to kill the job, shut down MySQL, reconfigure the data directory and the socket location, move the contents over to the new location, and restart MySQL. Here are the configuration changes:

1 2 3 4 5 6 7

# Source: /etc/my.cnf [mysqld] #datadir=/var/lib/mysql #socket=/var/lib/mysql/mysql.sock datadir=/prod/mysql_db socket=/prod/mysql_db/mysql.sock ...

Deployment

Before this, I used to write shell scripts that set the JARS required by Hadoop and my application in the classpath, and then called Java. When I was doing this, I discovered that you can use the $HADOOP_HOME/bin/hadoop to call your custom Map-Reduce tasks as well, so I decided to use that.

However, I needed to set a few custom JAR files that Hadoop did not have (or need) in its classpath. I was using commons-codec which provided me implementations of Soundex and Metaphone, and I was writing to a MySQL database for which I needed the JDBC driver JAR, plus a few others for functionality I was too lazy to implement on my own.

There are two ways to supply these extra JAR files to the bin/hadoop script. One is by specifying their paths in the -libjars parameter. I thought this was nice, but it didn't work for me - for some reason it could not see the parameters I was passing to my Map-Reduce job via the command line. The second way is to package your custom JARs in the lib subdirectory of your application's JAR file, a so-called fat jar. The fat JAR approach was the one I took, creating it using the simple Ant target shown below:

1 2 3 4 5 6 7 8 9 10 11 12 13

<target name="fatjar" depends="compile" description="Build JAR to run in Hadoop"> <mkdir dir="${maven.build.output}/lib"/> <copy todir="${maven.build.output}/lib"> <fileset dir="${custom.jars.dir}"> <include name="commons-lang3-3.0.1.jar"/> <include name="commons-codec-1.3.jar"/> <include name="mysql-connector-java-5.0.5-bin.jar"/> ... </fileset> </copy> <jar jarfile="${maven.build.directory}/${maven.build.final.name}-fatjar.jar" basedir="${maven.build.output}" excludes="**/package.html"/> </target>

Once this is done, the script to run either job is quite simple. I show the cscript to run the BigramCounter below, simply replace with UnigramCounter for the other one.

1 2 3 4 5 6

#!/bin/bash # Source: bin/bigram_counter.sh HADOOP_HOME=/opt/hadoop-1.0.1 $HADOOP_HOME/bin/hadoop fs -rmr /prod/hadoop/dummy $HADOOP_HOME/bin/hadoop jar /path/to/my-fatjar.jar \ com.mycompany.spell3.train.BigramCounter $*

To run this script from the command line:

1 2	hduser@bigmac:spell3$ nohup ./bigram_counter.sh /prod/hadoop/spell \ spelldb spelluser spelluser &

Job Killing

I needed to kill the job midway multiple times, either because I discovered I had goofed on some programming issue (incorrect database column names, etc) and the job would start throwing all kinds of exceptions down the line, or because (as mentioned previously), MySQL ran out of disk space. To do this, you need to use bin/hadoops job -kill command.

1 2 3 4

hduser@bigmac:hadoop-1.0.1$ # list out running jobs hduser@bigmac:hadoop-1.0.1$ bin/hadoop job -list hduser@bigmac:hadoop-1.0.1$ # kill specific job hduser@bigmac:hadoop-1.0.1$ bin/hadoop job -kill ${job-id}

Even I had enough sense to not do a kill -9 on the hadoop daemon itself, but there was one time when I did a stop-all.sh and ended up having to throw away all my data because Hadoop got all choked up.

Another little tip is to avoid throwing exceptions from your Mapper or Reducer. A better option is to log it. This is true for any batch job, of course, but I once had one of the jobs fail after about 2 days of processing because of too many exceptions thrown by the Reducer. In the code above, I just used a System.ot.println() to log SQLExceptions if they occur, but its better to use a real logger.

So anyway, after about a week and a half of processing (including all sorts of silly but expensive mistakes), I ended up with approximately 400 million unigrams and 600 million bigrams in the database. Now to figure out how to actually use this information :-).

Update - 2012-04-09: I had a bug in my bigram generation code, which caused bad results, so I reran it. This time the job failed two times in a row, caused by (I suspect) extremely high loads on the MySQL database server. The first time I discovered that the mysql.sock file disappeared, so I terminated the job manually. The second time I found that the mysql.sock file would disappear and then reappear after a while once the load came back down (this is the only place I have found another mention of this) - however, ultimately this job failed as well. I ended up writing the bigrams and counts to text files in HDFS and the job completed in a fraction of the time it took before. So another lesson learned - avoid writing out to external datastores from within Hadoop,

Multithreaded TGNI Concept Loader

Sometime back, I mentioned that I tried to load up our taxonomy (with about 1 million medical concepts), into TGNI's Lucene and Neo4J datastores, and the process took 3 weeks to complete (on my 2 CPU desktop at work, as a single threaded process). I've been meaning to see if I could speed it up, but the data was adequate for most of the experiments I was doing, so I did not have enough incentive. Until about 4 weeks ago, when I discovered that I had inadverdently pulled in retired and experimental concepts and that they were interfering with the quality of my output.

My initial plan was to convert the loading process into a Map-Reduce job with Hadoop, but I would have to server-ize Lucene and Neo4j (ie, using SOLR and Neo4j's REST API), and the prospect of having to start up 3 servers to test the application seemed a bit daunting, so I scrapped that idea in favor of just multi-threading the loading application. Although, in retrospect, that would have worked equally well (in terms of effort involved to implement) and would have been more scalable (in terms of the hardware requirements - its far easier to get a bank of low-powered servers than it is to get a single high-powered server).

In this post, I describe the somewhat convoluted process that led to a successful multi-threaded loader implementation, hoping that somewhere in this, there are lessons for people (like myself and possibly a vast majority of Java programmers) to whom writing non-trivial multithreaded apps is like buying a car, ie, something you have to do only once every say 5-7 years.

To provide some context, here is what the flow in my original (single threaded) loader looked like. The code would loop through a bunch of tables in an Oracle database and build concept objects out of it, then send the object to a node service, which consisted of a graph service and an index service. The concept would be added to the Neo4j graph database (and get a node ID in the process), then it would be sent to the index service, which would pass it through the UIMA/Lucene analyzer chain to create an entry (heavily augmented with attributes) in the Lucene index for each name (primary, qualified, synonyms) associated witht he concept.

My first implementation was to build a list of OIDs from the Oracle database, then spawn a fixed size thread pool using Java's ExecutorService. Each thread would then build a TConcept object, write to Neo4j, normalize the names and add them (as distinct entities) to the MySQL database. This would run through about 3,000 concepts before hanging. Thinking that perhaps it was something to do with the way I had integrated UIMA with Lucene analyzers, I broke them apart so the UIMA Analysis Engine (AE) would annotate each input name, then break them apart into (potentially) multiple strings, then feed them in, one by one, into the Lucene analyzer chain consisting of streaming Lucene only components (keyword attribute aware LowerCaseFilter, StopFilter and PorterStemFilter).

While I was doing this, I decided to switch out Lucene and use MySQL instead. I was pre-normalizing the names anyway, and I needed to match normalized versions of my input against normalized versions of the concept names. Using Lucene wasn't buying me anything - it was actually hurting because it would match partial strings, and I was having to write code to prevent that.

However, the pipeline would still hang at around the same point. I remembered that I had used Jetlang some time back, and decided to see if modeling it as a Jetlang actor would help. This version ran through about 70,000 concepts before it hung. While I was running this version, I noticed that the CPUs ran a lot cooler (using top and looking at the user CPU consumed) with the Jetlang version compared to my original multithreaded version.

At that point I realized that each of my threads in my original version was creating its own version of the UIMA AE, Lucene Analyzer and database Connection objects for each concept. Since Jetlang uses the Actor model, its threads were basically mini-servers that looped in a read-execute loop.

In an attempt to keep the code mostly intact (I was trying to reuse code as far as possible), I factored out these resources into pools using Commons-Pool and replaced the constructor (and destructor) calls with calls to pool.borrowObject() and pool.returnObject(). This helped, in the sense that I noticed less CPU utilization, but the job would just mysteriously block at around the same point, ie, no movement in the logs, top showing no activity except in one or two CPUs.

Digging deeper, I found that chemical names were being caught by my semantic hyphen transformation pattern (meant to expand hyphenated words into two word and single word tokens), and were generating thousands of synonyms for them.

At the same time, I realized that I could dispense with the pools altogether by modeling my threads as mini-servers (with a for(;;) loop breakable with a poison pill message) and giving each thread its own copy of an UIMA AE, Analyzer, Oracle and MySQL Connection objects. Neo4j allows only a single connection to the database, but is thread-safe, so I wrapped the connection in a singleton and gave each mini-server a reference to the singleton.

For chemical names, I put in an additional AE and changed the flow so if a string (or part of it) was already annotated, a downstream AE will not attempt to annotate it. However, just in case there were other wierd patterns lurking in the input, I wanted to be able to terminate the normalization process (and not process the concept) if it took "too long" to execute, so it did not hold up other concepts that could be processed.

With all these requirements, I ended up modeling the job in three levels - the manager which instantiates everything and creates a queue of input ids to process, a pool of worker threads which are mini-servers and which have their own instances of expensive resources, and normalization tasks, which are instantiated as callable futures from within the worker threads, and which timeout after a configurable amount of time (default 1s), and cause the UIMA CAS (an expensive resource that should be destroyed according to the UIMA docs) to be released and the AE rebuilt with a new CAS when that happens.

Here's the code (with the application specific stuff elided to keep it short, since it adds nothing to the discussion).

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440

// Source: src/main/java/com/mycompany/tgni/loader/ConceptLoadManager.java package com.mycompany.tgni.loader; import java.io.File; import java.io.Reader; import java.math.BigDecimal; import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import opennlp.tools.util.Pair; import org.apache.commons.collections15.CollectionUtils; import org.apache.commons.lang.StringUtils; import org.apache.lucene.analysis.Analyzer; import org.apache.lucene.analysis.TokenStream; import org.apache.lucene.analysis.core.LowerCaseFilter; import org.apache.lucene.analysis.en.PorterStemFilter; import org.apache.lucene.analysis.standard.StandardTokenizer; import org.apache.lucene.util.Version; import org.apache.uima.analysis_engine.AnalysisEngine; import org.apache.uima.jcas.JCas; import org.neo4j.graphdb.GraphDatabaseService; import org.neo4j.graphdb.Node; import org.neo4j.graphdb.Transaction; import org.slf4j.Logger; import org.slf4j.LoggerFactory; //import org.springframework.util.StopWatch; import com.mycompany.tgni.beans.TConcept; import com.mycompany.tgni.lucene.StopFilter; import com.mycompany.tgni.neo4j.GraphInstance; import com.mycompany.tgni.neo4j.JsonUtils; import com.mycompany.tgni.neo4j.NameNormalizer; import com.mycompany.tgni.uima.utils.UimaUtils; public class ConceptLoadManager { private final Logger logger = LoggerFactory.getLogger(getClass()); private static final int NUM_WORKERS = Math.round(1.4F * Runtime.getRuntime().availableProcessors()); private static final long TASK_TIMEOUT_MILLIS = 1000L; private static final CountDownLatch LATCH = new CountDownLatch(NUM_WORKERS); private static final BlockingQueue<Integer> QUEUE = new LinkedBlockingQueue<Integer>(); // oracle queries private static final String LIST_OIDS_SQL = "..."; private static final String GET_HEAD_SQL = "..."; private static final String GET_PNAMES_SQL = "..."; private static final String GET_SYNS_SQL = "..."; private static final String GET_STY_SQL = "..."; // mysql queries private static final String ADD_NAME_SQL = "insert into oid_name (" + "oid, name, pri) " + "values (?,?,?)"; private static final String ADD_NID_SQL = "insert into oid_nid (oid, nid) values (?, ?)"; public static void main(String[] args) throws Exception { // extract parameters from command line if (args.length != 5) { System.out.println("Usage: ConceptLoadManager " + "/path/to/graph/dir /path/to/mysql-properties " + "/path/to/stopwords/file /path/to/ae/descriptor " + "/path/to/oracle-properties"); System.exit(-1); } // Initialize manager ConceptLoadManager manager = new ConceptLoadManager(); final GraphInstance neo4jConn = new GraphInstance(args[0]); final String mysqlProps = args[1]; final Set<?> stopwords = StopFilter.makeStopSet( Version.LUCENE_40, new File(args[2])); final String aeDescriptor = args[3]; final String oraProps = args[4]; // seed input queue manager.seed(oraProps); // add poison pills for (int i = 0; i < NUM_WORKERS; i++) { try { QUEUE.put(-1); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } // set up worker threads ExecutorService workerPool = Executors.newFixedThreadPool(NUM_WORKERS); for (int i = 0; i < NUM_WORKERS; i++) { ConceptLoadWorker worker = new ConceptLoadManager().new ConceptLoadWorker( i, mysqlProps, stopwords, aeDescriptor, oraProps, neo4jConn); workerPool.execute(worker); } // wait for all tasks to process, then shutdown workerPool.shutdown(); try { LATCH.await(); } catch (InterruptedException e) { /* NOOP */ } neo4jConn.destroy(); workerPool.awaitTermination(1000L, TimeUnit.MILLISECONDS); } private void seed(String oraProps) { List<Integer> oids = new ArrayList<Integer>(); Connection conn = null; PreparedStatement ps = null; ResultSet rs = null; try { conn = DbConnectionUtils.getConnection(oraProps); ps = conn.prepareStatement(LIST_OIDS_SQL); rs = ps.executeQuery(); while (rs.next()) { QUEUE.put(rs.getInt(1)); } } catch (Exception e) { logger.warn("Can't generate OIDs to process", e); } finally { DbConnectionUtils.closeResultSet(rs); DbConnectionUtils.closePreparedStatement(ps); DbConnectionUtils.closeConnection(conn); } } /////////////// Worker Class /////////////////// private class ConceptLoadWorker implements Runnable { private int workerId; private AtomicInteger count; private int totalTasks; private Set<?> stopwords; private String mysqlProps; private String aeDescriptor; private String oraProps; private GraphInstance neo4jConn; private Connection mysqlConn; private PreparedStatement psAddNames, psAddNid; private Connection oraConn; private PreparedStatement psGetHead, psGetNames, psGetSyns, psGetSty; private AnalysisEngine ae; private JCas jcas; private Analyzer analyzer; public ConceptLoadWorker(int workerId, String mysqlProps, Set<?> stopwords, String aeDescriptor, String oraProps, GraphInstance neo4jConn) { this.workerId = workerId; this.count = new AtomicInteger(0); this.totalTasks = QUEUE.size(); this.mysqlProps = mysqlProps; this.stopwords = stopwords; this.aeDescriptor = aeDescriptor; this.oraProps = oraProps; this.neo4jConn = neo4jConn; } @Override public void run() { try { initWorker(); ExecutorService taskExec = Executors.newSingleThreadExecutor(); for (;;) { Integer oid = QUEUE.take(); if (oid < 0) { break; } int curr = count.incrementAndGet(); // load the concept by OID from oracle TConcept concept = null; try { concept = loadConcept(oid); } catch (SQLException e) { logger.warn("Exception retrieving concet (OID:" + oid + ")", e); continue; } // normalize names using UIMA/Lucene chains. This is // a slow process so we want to time this out if it // takes too long. In that case, the node/oid mapping // will not be written out into Neo4J. NameNormalizer normalizer = new NameNormalizer(ae, analyzer, jcas); NameNormalizerTask task = new NameNormalizerTask( concept, normalizer); Future<List<Pair<String,Boolean>>> futureResult = taskExec.submit(task); List<Pair<String,Boolean>> result = null; try { result = futureResult.get( TASK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); } catch (ExecutionException e) { logger.warn("Task (OID:" + oid + ") skipped", e); reinitWorker(); continue; } catch (TimeoutException e) { futureResult.cancel(true); logger.warn("Task (OID:" + oid + ") timed out", e); reinitWorker(); continue; } try { // add the OID-Name mappings to MySQL addNames(oid, result); // add the OID-NID mapping to Neo4j writeNodeConceptMapping(concept); } catch (Exception e) { logger.warn("Exception persisting concept (OID:" + oid + ")", e); continue; } // report on progress if (curr % 100 == 0) { logger.info("Worker " + workerId + " processed (" + curr + "/" + totalTasks + ") OIDs"); } } taskExec.shutdownNow(); } catch (InterruptedException e) { logger.error("Worker:" + workerId + " Interrupted", e); } catch (Exception e) { logger.error("Worker:" + workerId + " threw exception", e); } finally { destroyWorker(); LATCH.countDown(); } } private TConcept loadConcept(Integer oid) throws SQLException { TConcept concept = new TConcept(); // bunch of SQLs run against Oracle database to populate // the concept ... return concept; } private void addNames(Integer oid, List<Pair<String, Boolean>> names) throws SQLException { if (names == null) return; try { psAddNames.clearBatch(); for (Pair<String,Boolean> name : names) { if (StringUtils.length(StringUtils.trim(name.a)) > 255) { continue; } psAddNames.setInt(1, oid); psAddNames.setString(2, name.a); psAddNames.setString(3, name.b ? "T" : "F"); psAddNames.addBatch(); } psAddNames.executeBatch(); mysqlConn.commit(); } catch (SQLException e) { mysqlConn.rollback(); throw e; } } private void writeNodeConceptMapping(TConcept concept) throws Exception { logger.info("Writing concept (OID=" + concept.getOid() + ")"); GraphDatabaseService graphService = neo4jConn.getInstance(); Transaction tx = graphService.beginTx(); try { // update neo4j Node node = graphService.createNode(); concept.setNid(node.getId()); node.setProperty("oid", concept.getOid()); node.setProperty("pname", concept.getPname()); node.setProperty("qname", concept.getQname()); node.setProperty("synonyms", JsonUtils.listToString(concept.getSynonyms())); node.setProperty("stycodes", JsonUtils.mapToString(concept.getStycodes())); node.setProperty("stygrp", StringUtils.isEmpty( concept.getStygrp()) ? "UNKNOWN" : concept.getStygrp()); node.setProperty("mrank", concept.getMrank()); node.setProperty("arank", concept.getArank()); node.setProperty("tid", concept.getTid()); // update mysql psAddNid.setInt(1, concept.getOid()); psAddNid.setLong(2, concept.getNid()); psAddNid.executeUpdate(); mysqlConn.commit(); tx.success(); } catch (Exception e) { mysqlConn.rollback(); tx.failure(); logger.info("Exception writing mapping (OID=" + concept.getOid() + ")"); throw e; } finally { tx.finish(); } } private void initWorker() throws Exception { logger.info("Worker:" + workerId + " init"); // mysql this.mysqlConn = DbConnectionUtils.getConnection(mysqlProps); this.mysqlConn.setAutoCommit(false); this.psAddNames = mysqlConn.prepareStatement(ADD_NAME_SQL); this.psAddNid = mysqlConn.prepareStatement(ADD_NID_SQL); // oracle this.oraConn = DbConnectionUtils.getConnection(oraProps); this.psGetHead = oraConn.prepareStatement(GET_HEAD_SQL); this.psGetNames = oraConn.prepareStatement(GET_PNAMES_SQL); this.psGetSyns = oraConn.prepareStatement(GET_SYNS_SQL); this.psGetSty = oraConn.prepareStatement(GET_STY_SQL); // uima/lucene this.ae = UimaUtils.getAE(aeDescriptor, null); this.analyzer = getAnalyzer(stopwords); this.jcas = ae.newJCas(); } private void destroyWorker() { // mysql DbConnectionUtils.closePreparedStatement(psAddNames); DbConnectionUtils.closePreparedStatement(psAddNid); DbConnectionUtils.closeConnection(this.mysqlConn); // oracle DbConnectionUtils.closePreparedStatement(psGetHead); DbConnectionUtils.closePreparedStatement(psGetNames); DbConnectionUtils.closePreparedStatement(psGetSyns); DbConnectionUtils.closePreparedStatement(psGetSty); DbConnectionUtils.closeConnection(this.oraConn); // uima/lucene this.ae.destroy(); this.analyzer.close(); this.jcas.release(); this.jcas.reset(); } private void reinitWorker() throws Exception { this.ae.destroy(); this.analyzer.close(); this.jcas.release(); this.jcas.reset(); this.ae = UimaUtils.getAE(aeDescriptor, null); this.analyzer = getAnalyzer(stopwords); this.jcas = ae.newJCas(); } private Analyzer getAnalyzer(final Set<?> stopwords) { return new Analyzer() { @Override public TokenStream tokenStream(String fieldName, Reader reader) { TokenStream input = new StandardTokenizer(Version.LUCENE_40, reader); input = new LowerCaseFilter(Version.LUCENE_40, input); input = new StopFilter(Version.LUCENE_40, input, stopwords);; input = new PorterStemFilter(input); return input; } }; } } ///////////////// Task class //////////////// private class NameNormalizerTask implements Callable<List<Pair<String,Boolean>>> { private TConcept concept; private NameNormalizer normalizer; public NameNormalizerTask(TConcept concept, NameNormalizer normalizer) { this.concept = concept; this.normalizer = normalizer; } @Override public List<Pair<String,Boolean>> call() throws Exception { logger.info("Executing task (OID:" + concept.getOid() + ")"); Set<String> uniques = new HashSet<String>(); Set<String> normalizedUniques = new HashSet<String>(); List<Pair<String,Boolean>> results = new ArrayList<Pair<String,Boolean>>(); String pname = concept.getPname(); if (StringUtils.isNotEmpty(pname) && (! uniques.contains(pname))) { List<String> normalized = normalizer.normalize(pname); uniques.add(pname); normalizedUniques.addAll(normalized); } String qname = concept.getQname(); if (StringUtils.isNotEmpty(qname) && (! uniques.contains(qname))) { List<String> normalized = normalizer.normalize(qname); uniques.add(qname); normalizedUniques.addAll(normalized); } for (String normalizedUnique : normalizedUniques) { results.add(new Pair<String,Boolean>(normalizedUnique, true)); } Set<String> normalizedUniqueSyns = new HashSet<String>(); normalizedUniqueSyns.addAll(normalizedUniques); List<String> syns = concept.getSynonyms(); for (String syn : syns) { if (StringUtils.isNotEmpty(syn) && (! uniques.contains(syn))) { List<String> normalizedSyn = normalizer.normalize(syn); uniques.add(syn); normalizedUniqueSyns.addAll(normalizedSyn); } } Collection<String> normalizedSyns = CollectionUtils.subtract( normalizedUniques, normalizedUniqueSyns); for (String normalizedSyn : normalizedSyns) { results.add(new Pair<String,Boolean>(normalizedSyn, false)); } return results; } } }

Since the worker threads were doing a combination of IO (reading from the Oracle database and writing to MySQL and Neo4j) and CPU bound work (normalizing with the UIMA AE and Lucene Analyzers), I ran some timings on a small sample of 1000 concepts and found that it spent approximately 30% of its time doing IO. So based on the formula in Java Concurrency in Practice book:

1	num_threads = num_cpus * target_cpu_utilization * (1 + wait/compute)

I set the number of worker threads to 22 on my 16 CPU machine. During the run, I noticed that the load average was between 3-4 (which is quite low for a 16 CPU box) and the user CPU utilization percentages hovered in the 2-3% mark on most but 2-3 CPUs, which showed around 40-50% utilization. So there is probably still some room for increasing the number of worker threads. Here is a screenshot of top while the program is running.

With 22 threads, the job finished in a very acceptable time of about 1.5 hours, with 88 concepts timing out. I plan to look at those concepts to see if I can uncover patterns that would lead to the creation of some more pre-emptive AEs in the future.

Meanwhile, I hope I'll remember this stuff the next time I need to build one of these things :-). Its almost Christmas, so for those of you who celebrate it, heres wishing you a very Merry Christmas!