databasemigration.java

一个java的ETL工具

import java.util.HashMap;
import java.util.Hashtable;
import java.util.HashSet;
import java.util.Vector;
import java.util.ArrayList;
import java.util.StringTokenizer;
import java.util.Arrays;
import java.util.GregorianCalendar;

import java.io.Reader;

import org.apache.xerces.parsers.DOMParser;
import org.xml.sax.InputSource;
import org.w3c.dom.Attr;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.NamedNodeMap;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;

import java.sql.DriverManager;
import java.sql.DatabaseMetaData;
import java.sql.Connection;
import java.sql.Statement;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
import java.sql.Types;

import Migration;
import Transfer;
import Field;
import Rule;
import Condition;
import Comparison;
import Action;
import UpdateFallbackRule;

public class DatabaseMigration
{
	// main arguments stuff

	private String xmlFileURL;
	private String operation = "TEST";
	private boolean debug = false;
	private String errorLevel = "IGNORE";

	// xml parsing stuff

	private Element xmlRoot;

	// connection stuff

	boolean connected = false;

	private String sourceDB_Driver = "";
	private String sourceDB_URL = "";
	private String sourceDB_Username = "";
	private String sourceDB_Password = "";
	private Connection sourceConnection = null;

	private String targetDB_Driver = "";
	private String targetDB_URL = "";
	private String targetDB_Username = "";
	private String targetDB_Password = "";
	private Connection targetConnection = null;

	// build migrations stuff

	ArrayList migration_AL = new ArrayList();

	// utility stuff

	private String tempString = "";
	private Vector tempVector = new Vector();
	private ArrayList tempArrayList = new ArrayList();
	private HashMap tempHashMap = new HashMap();

	private NodeList tempNodeList = null;
	private Node tempNode = null;
	private Element tempElement;


	// Command Line Main Method

	public static void main(String[] args)
	{
System.out.println("main:  start");
		String arg_xmlFileURL = "";
		String arg_operation = "TEST";
		boolean arg_debug = true;

		String usage = "usage:  java DatabaseMigration XML_FileURL run|test debug|nodebug";

		if (args.length >= 1)
			arg_xmlFileURL = args[0];
		else
		{
			System.out.println("Please specify XML File URL");
			System.out.println("\n");
			System.out.println(usage);
			System.exit(0);
		}

		if (args.length >= 2)
		{
			arg_operation = args[1];
			arg_operation = arg_operation.toUpperCase();

			if (!arg_operation.equals("RUN") && !arg_operation.equals("TEST")) arg_operation = "TEST";
		}

		if (args.length >=3)
		{
			String arg_debugString = args[2];
			arg_debugString = arg_debugString.toUpperCase();
			if (arg_debugString.equals("NODEBUG")) arg_debug = false;
		}	


//		DatabaseMigration dbMigration = new DatabaseMigration(arg_xmlFileURL, arg_operation, arg_debug);
		
// temporary - can't figure out how to send a second command line argument via KAWA
		DatabaseMigration dbMigration = new DatabaseMigration(arg_xmlFileURL, "RUN", false);

		dbMigration.buildRunParameters();
		dbMigration.connectToDatabases();
		dbMigration.buildMigrations();
		try
		{
			dbMigration.migrate();
		}
		catch (Exception e)
		{
			System.out.println(e);
		}	
	}

	// Constructors

	public DatabaseMigration(String _xmlFileURL, String _operation, boolean _debug)
	{
		xmlFileURL = _xmlFileURL;
		operation = _operation;
		debug = _debug;

		System.out.println("XML File = " + xmlFileURL);
		System.out.println("Operation = " + operation);
		System.out.println("Debug Mode = " + debug);

		xmlRoot = getRoot(xmlFileURL);
	}

	//  XML Parsing Methods

	private Element getRoot(String sentXML_URL)
	{
System.out.println("getRoot:  start");
		Element root = null;

		try
		{
			DOMParser dp = new DOMParser();
			dp.parse(sentXML_URL);
			org.w3c.dom.Document doc = dp.getDocument();
			root = doc.getDocumentElement();
			root.normalize();
		}
		catch (Exception e)
		{
			System.out.println("Error parsing XML document");
			System.exit(0);
		}

System.out.println("getRoot:  end");
		return root;
	}


	private void buildMigrations()
	{
		migration_AL.clear();

		NodeList migration_NL = xmlRoot.getElementsByTagName("TABLE_MIGRATION");
System.out.println("migrations found = " + migration_NL.getLength());		
		for (int i=0; i<migration_NL.getLength(); i++)
		{
System.out.println("buildMigrations:  start");
			Element tableMigration_EL = (Element)migration_NL.item(i);

			String migrationName = tableMigration_EL.getAttribute("NAME");
			String migrationSourceTable = tableMigration_EL.getAttribute("TABLE_SOURCE");
			String migrationTargetTable = tableMigration_EL.getAttribute("TABLE_TARGET");

			Element fields_EL = (Element)tableMigration_EL.getElementsByTagName("FIELDS").item(0);

			// Build List of Source Table Used Fields

System.out.println("buildMigrations:source fields  start");
			Element sourceFields_EL = (Element)fields_EL.getElementsByTagName("SOURCE_FIELDS").item(0);
			ArrayList sourceField_AL = new ArrayList();
			tempNodeList = sourceFields_EL.getElementsByTagName("FIELD_DEFINITION");
			for (int j=0; j<tempNodeList.getLength(); j++)
			{
				String id = ((Element)tempNodeList.item(j)).getAttribute("ID");
				String nameSQL = ((Element)tempNodeList.item(j)).getAttribute("NAME");
				String typeSQL = ((Element)tempNodeList.item(j)).getAttribute("SQL_TYPE");
System.out.println(id + "=" + nameSQL + "(" + typeSQL + ")");

				sourceField_AL.add(new Field(id, nameSQL, typeSQL));
			}	
			sourceField_AL.trimToSize();
				
			// Build List of Target Table Used Fields
			
System.out.println("buildMigrations:target fields  start");
			Element targetFields_EL = (Element)fields_EL.getElementsByTagName("TARGET_FIELDS").item(0);
			ArrayList targetField_AL = new ArrayList();
			tempNodeList = targetFields_EL.getElementsByTagName("FIELD_DEFINITION");
			for (int j=0; j<tempNodeList.getLength(); j++)
			{
				String id = ((Element)tempNodeList.item(j)).getAttribute("ID");
				String nameSQL = ((Element)tempNodeList.item(j)).getAttribute("NAME");
				String typeSQL = ((Element)tempNodeList.item(j)).getAttribute("SQL_TYPE");
System.out.println(id + "=" + nameSQL + "(" + typeSQL + ")");

				targetField_AL.add(new Field(id, nameSQL, typeSQL));
			}	
			targetField_AL.trimToSize();
			
			// Build List of Temporary Variables 
				
System.out.println("buildMigrations:temporary variables  start");
			Element temporaryVariables_EL = (Element)fields_EL.getElementsByTagName("TEMPORARY_FIELDS").item(0);
			ArrayList temporaryVariable_AL = new ArrayList();
			tempNodeList = temporaryVariables_EL.getElementsByTagName("VARIABLE_DEFINITION");
			for (int j=0; j<tempNodeList.getLength(); j++)
			{
				String id = ((Element)tempNodeList.item(j)).getAttribute("ID");
				String typeSQL = ((Element)tempNodeList.item(j)).getAttribute("SQL_TYPE");
System.out.println(id + "=(" + typeSQL + ")");

				temporaryVariable_AL.add(new Variable(id, typeSQL));
			}	
			temporaryVariable_AL.trimToSize();
			
			// Build Transfers	
				
			ArrayList transfer_AL = new ArrayList();
			NodeList transfer_NL = tableMigration_EL.getElementsByTagName("INSERT_ROW_TRANSFER");
			for (int j=0; j<transfer_NL.getLength(); j++)
			{
System.out.println("buildMigrations:transfer  start");
				Element transfer_EL = (Element)transfer_NL.item(j);

				String description = transfer_EL.getAttribute("DESCRIPTION");

				//  build Condition

				Condition condition = null;
				NodeList condition_NL = transfer_EL.getElementsByTagName("CONDITION");
				if (condition_NL.getLength() == 1)
				{
System.out.println("buildMigrations:condition  start");
					Element condition_EL = (Element)condition_NL.item(0);  //  only one CONDITION allowed

					//   Build Comparisons
										
					ArrayList comparison_AL = new ArrayList();
					NodeList comparison_NL = condition_EL.getElementsByTagName("COMPARISON");
					for (int k=0; k<comparison_NL.getLength(); k++)
					{
System.out.println("buildMigrations:comparison  start");
						Element comparison_EL = (Element)comparison_NL.item(k);
						
						String operator = comparison_EL.getAttribute("OPERATOR");
						String relationToPrevious = comparison_EL.getAttribute("RELATION_TO_PREVIOUS");
						String firstOperand = ((Element)comparison_EL.getElementsByTagName("FIRST_OPERAND").item(0)).getFirstChild().getNodeValue();
						String secondOperand = ((Element)comparison_EL.getElementsByTagName("SECOND_OPERAND").item(0)).getFirstChild().getNodeValue();
					
						Comparison comparison = null;
						if (relationToPrevious == null)
							comparison = new Comparison(operator, firstOperand, secondOperand);
						else
							comparison = new Comparison(operator, relationToPrevious, firstOperand, secondOperand);
												
						comparison_AL.add(comparison);
					}
					comparison_AL.trimToSize();
					condition = new Condition(comparison_AL);	
				}
				
				//  Build Actions
								
				ArrayList action_AL = new ArrayList();
				NodeList action_NL = transfer_EL.getElementsByTagName("ACTION");
				for (int k=0; k<action_NL.getLength(); k++)
				{
System.out.println("buildMigrations:action  start");
					Element actionSubElement = (Element)action_NL.item(k).getChildNodes().item(1);
					
					Action action = null;
					if (actionSubElement.getNodeName().equals("PUSH"))
					{
						String fromId = actionSubElement.getElementsByTagName("FROM_ID").item(0).getFirstChild().getNodeValue();
						String toId = actionSubElement.getElementsByTagName("TO_ID").item(0).getFirstChild().getNodeValue();

						action = new Push(Action.PUSH, fromId, toId);
					}
					else if (actionSubElement.getNodeName().equals("STRING_REPLACE"))
					{
						String fromId = actionSubElement.getElementsByTagName("FROM_ID").item(0).getFirstChild().getNodeValue();
						String toId = actionSubElement.getElementsByTagName("TO_ID").item(0).getFirstChild().getNodeValue();

						String fromText = actionSubElement.getAttribute("FROM_TEXT");
						String toText = actionSubElement.getAttribute("TO_TEXT");

						action = new StringReplace(Action.STRING_REPLACE, fromId, fromText, toId, toText);
					}
					else if (actionSubElement.getNodeName().equals("STRING_SPLIT"))
					{
						String delimiter = actionSubElement.getAttribute("DELIMITER");
						String fromId = actionSubElement.getElementsByTagName("FROM_ID").item(0).getFirstChild().getNodeValue();
						ArrayList toId_AL = new ArrayList();
						NodeList toId_NL = ((Element)actionSubElement.getElementsByTagName("IDS_IN_ORDER").item(0)).getElementsByTagName("TO_ID");	
						for (int l=0; l<toId_NL.getLength(); l++)
						{
							String toId = ((Element)toId_NL.item(l)).getFirstChild().getNodeValue();
							toId_AL.add(toId);
						}
						toId_AL.trimToSize();

						action = new StringSplit(Action.STRING_SPLIT, fromId, toId_AL, delimiter);
					}
					else if (actionSubElement.getNodeName().equals("STRING_MERGE"))
					{
						String delimiter = actionSubElement.getAttribute("DELIMITER");
						ArrayList fromId_AL = new ArrayList();
						NodeList fromId_NL = ((Element)actionSubElement.getElementsByTagName("IDS_IN_ORDER").item(0)).getElementsByTagName("FROM_ID");	
						for (int l=0; l<fromId_NL.getLength(); l++)
						{
							String fromId = fromId_NL.item(l).getFirstChild().getNodeValue();
							fromId_AL.add(fromId);
						}
						fromId_AL.trimToSize();
						String toId = actionSubElement.getElementsByTagName("TO_ID").item(0).getFirstChild().getNodeValue();

						action = new StringMerge(Action.STRING_MERGE, fromId_AL, toId, delimiter);
					}
					else if (actionSubElement.getNodeName().equals("CONVERT"))
					{
						String fromId = actionSubElement.getElementsByTagName("FROM_ID").item(0).getFirstChild().getNodeValue();
						String toId = actionSubElement.getElementsByTagName("TO_ID").item(0).getFirstChild().getNodeValue();

						action = new Convert(Action.CONVERT, fromId, toId);
					}
					action_AL.add(action);
				}
				action_AL.trimToSize();

				Transfer transfer = new Transfer(description, condition, action_AL);
				transfer_AL.add(transfer);
			}
			
			//  Build Update Fallback Rule

			UpdateFallbackRule updateFallbackRule = new UpdateFallbackRule();
			NodeList updateFallbackRule_NL = tableMigration_EL.getElementsByTagName("UPDATE_FALLBACK_RULE");
			if (updateFallbackRule_NL.getLength() == 1)
			{
System.out.println("buildMigrations:UpdateFallbackRule  start");
				Element updateFallbackRule_EL = (Element)updateFallbackRule_NL.item(0);  //  only one CONDITION allowed

				//   Build Target Updates
									
				ArrayList targetUpdateId_AL = new ArrayList();
				NodeList targetUpdateId_NL = updateFallbackRule_EL.getElementsByTagName("TARGET_UPDATE_ID");
				for (int j=0; j<targetUpdateId_NL.getLength(); j++)
				{
					Element targetUpdateId_EL = (Element)targetUpdateId_NL.item(j);
					
					String fieldId = targetUpdateId_EL.getFirstChild().getNodeValue();
					String mode = targetUpdateId_EL.getAttribute("MODE");
				
					TargetUpdateField updateField = new TargetUpdateField(fieldId, mode);
											
					targetUpdateId_AL.add(updateField);
				}
				targetUpdateId_AL.trimToSize();
				updateFallbackRule = new UpdateFallbackRule(targetUpdateId_AL);
System.out.println("buildMigrations:UpdateFallbackRule  end");
			}
			
			Migration migration = new Migration(migrationName, migrationSourceTable, migrationTargetTable, sourceField_AL, targetField_AL, temporaryVariable_AL, transfer_AL, updateFallbackRule);
			migration_AL.add(migration);

			if (debug || operation.equals("TEST")) System.out.println(migration);			
		}
		migration_AL.trimToSize();
	}

	//  Build Run Parameters
	
	private void buildRunParameters()
	{
		tempElement = (Element)xmlRoot.getElementsByTagName("RUN_PARAMETERS").item(0);  // there is one and only one named CONNECTION Element
		errorLevel = tempElement.getAttribute("ERROR_LEVEL");
	}
	
	//  Database Connection

	private void connectToDatabases()
	{
System.out.println("connectToDatabases:  start");
		if (connected == true) return;

		connected = false;