I have always wanted to duplicate how GoldenGate handles batch failures. In GoldenGate, you can set the parameter “batchsql BATCHTRANSOPS 2000”, for example, to send 2000 statements at a time to the database for execution. In general, this is far faster than sending them one at a time. If an error is encountered during batch execution, such as a duplicate key violation, the statements are rolled back and reprocessed serially. This is really useful if you don’t expect many errors. If an error does occur, the statements that can be successfully processed are executed, and the exceptions can be logged to a file.<\/p>\n
I wanted to do the same with JDBC, and came up with what is below. Roughly, this reads lines from a file and stores them in an ArrayList. Once the list has reached a configurable size, each is sent in a batch to the database. if an exception is thrown, the batch is rolled back and the ArrayList is resubmitted to a method that processes each statement serially. Any exceptions in this phase are logged, and the remainder that can be processed are executed.<\/p>\n
This class has some additional elements such as key processing, but for this post, the emphasis is on the batch processing algorithm.<\/p>\n
\r\nimport java.io.*;\r\nimport java.text.*;\r\nimport java.util.*;\r\nimport java.sql.*;\r\n\r\npublic class loadData {\r\n\r\n static Connection conn = null;\r\n static PreparedStatement pst = null;\r\n static ArrayList rows = new ArrayList();\r\n static ResultSet rst = null;\r\n static ResultSetMetaData rsmd = null;\r\n static DateFormat df = new SimpleDateFormat(\"MM\/dd\/yyyy H:m:s\");\r\n\r\n public static void main (String args[]) throws Exception {\r\n\r\n PGPFileProcessor pgpfp = new PGPFileProcessor();\r\n pgpfp.setInputFileName(args[0]);\r\n String output = null;\r\n pgpfp.setSecretKeyFileName(\"secring.gpg\");\r\n if (args[0].indexOf(\".gpg\") > 0)\r\n output = args[0].replace(\".gpg\",\"\");\r\n else\r\n output = args[0] + \".txt\";\r\n\r\n pgpfp.setOutputFileName(output);\r\n pgpfp.setPassphrase(\"****\");\r\n pgpfp.decrypt();\r\n\r\n BufferedReader bfr = new BufferedReader(new FileReader(output));\r\n\r\n conn = DriverManager.getConnection(\"jdbc:oracle:thin:user\/pwd@dbhost:1521\/dbservice\");\r\n String table = args[1];\r\n System.out.println(table);\r\n conn.setAutoCommit(false);\r\n rst = conn.createStatement().executeQuery(\"select * from \" + table);\r\n rsmd = rst.getMetaData();\r\n String insert = \"insert into \" + table + \" values(\";\r\n for (int i = 1; i <= rsmd.getColumnCount(); i++) {\r\n if (i == rsmd.getColumnCount()) {\r\n insert = insert + \"?)\";\r\n }\r\n else {\r\n insert = insert + \"?,\";\r\n }\r\n }\r\n pst = conn.prepareStatement(insert);\r\n int l = 0;\r\n String s;\r\n int cnt = 0;\r\n while((s = bfr.readLine()) != null) {\r\n if (cnt > 0) {\r\n int j = 0;\r\n String[] st2 = s.split(\"\\\\|\",-1);\r\n if (st2.length == rsmd.getColumnCount()) {\r\n rows.add(st2);\r\n }\r\n else {\r\n System.out.println(new java.util.Date() + \"\\tFAILURE: Column number mismatch \" + s);\r\n }\r\n if (l++ % 10000 == 0) {\r\n processBatch();\r\n try {\r\n pst.executeBatch();\r\n System.out.println(\"batch successful\");\r\n conn.commit();\r\n System.out.println(new java.util.Date() + \"\\tinserted \" + l + \" rows.\");\r\n rows.clear();\r\n }\r\n catch (Exception e2) {\r\n System.out.println(e2.getMessage());\r\n conn.rollback();\r\n processRowByRow();\r\n rows.clear();\r\n }\r\n }\r\n }\r\n cnt++;\r\n }\r\n processBatch();\r\n try {\r\n pst.executeBatch();\r\n conn.commit();\r\n System.out.println(new java.util.Date() + \"\\tinserted \" + l + \" rows.\");\r\n rows.clear();\r\n }\r\n catch (Exception e2) {\r\n System.out.println(e2.getMessage());\r\n conn.rollback();\r\n processRowByRow();\r\n rows.clear();\r\n }\r\n bfr.close();\r\n }\r\n\r\n static public void processRowByRow() throws Exception {\r\n System.out.println(\"row by row processing\");\r\n for (int m = 0; m < rows.size(); m++) {\r\n try {\r\n processRow(rows.get(m));\r\n pst.execute();\r\n \/\/rows.remove(m);\r\n conn.commit();\r\n }\r\n catch (Exception e) {\r\n System.out.println(rows.get(m)[0]);\r\n System.out.println(e.getMessage());\r\n }\r\n }\r\n }\r\n\r\n static public void processBatch() throws Exception {\r\n System.out.println(\"batch processing\");\r\n for (int m = 0; m < rows.size(); m++) {\r\n processRow(rows.get(m));\r\n pst.addBatch();\r\n }\r\n }\r\n\r\n static public void processRow(String[] s) throws Exception {\r\n int j = 0;\r\n for (int k = 0; k < s.length; k++) {\r\n j++;\r\n String tmp = s[k].replace(\"\\\"\",\"\");\r\n \/\/System.out.println(j + \" \" + tmp + \" \" + rsmd.getColumnTypeName(j) + \" \" + rsmd.getColumnName(j));\r\n if (rsmd.getColumnTypeName(j).equals(\"DATE\")) {\r\n if (tmp == null || tmp.equals(\"\"))\r\n pst.setNull(j, java.sql.Types.DATE);\r\n else\r\n pst.setDate(j,new java.sql.Date(df.parse(tmp).getTime()));\r\n }\r\n else if (rsmd.getColumnTypeName(j).equals(\"NUMBER\")) {\r\n if (tmp == null || tmp.equals(\"\"))\r\n pst.setNull(j, java.sql.Types.INTEGER);\r\n else\r\n pst.setDouble(j,Double.parseDouble(tmp));\r\n }\r\n else if (rsmd.getColumnTypeName(j).equals(\"VARCHAR2\")) {\r\n if (tmp == null)\r\n pst.setNull(j, java.sql.Types.VARCHAR);\r\n else\r\n pst.setString(j,tmp);\r\n }\r\n }\r\n }\r\n}\r\n<\/pre>\n","protected":false},"excerpt":{"rendered":"I have always wanted to duplicate how GoldenGate handles batch failures. In GoldenGate, you can set the parameter “batchsql BATCHTRANSOPS 2000”, for example, to send 2000 statements at a time to the database for execution. In general, this is far…<\/p>\n