Loading, computation and writing 500.000 rows in database

Question

first let me explain what I am trying to achieve here :

I am supposed to compute some results, using data from several tables, filtered by a run_id. One of these tables has about 500.000 rows per run_id, and 60 columns. All other tables have only a few thousand rows per run_id. I also have to write about 500.000 rows of data in a table with only 4 columns.

So, I first tried to load all these tables at once, which failed with an OutOfMemoryError because of the big table (called EPE from now on).

So, I decided to do some batch select to load the data from EPE. However, depending on the 'business' of the database, the whole process (load, compute, write) can take between 15 and 30 minutes.

Can I improve my code to make it faster ?

Here is my code, slightly reworked, so it might have some typos. Please let me know.

Main class :

@Component
public class Computation {

    public static final int MAX_RESULTS = 50000;
    public static final String SPECIAL = "207";

    Map<String, Curve> dfs;
    Map<String, Curve> pds;
    Map<String, LegalEntity> les;

    @Autowired
    @Qualifier("curveDAO")
    CurveAccess curveDAO;
    @Autowired
    @Qualifier("dealDAO")
    DealAccess dealDAO;
    @Autowired
    @Qualifier("entityDAO")
    EntityAccess entityDAO;
    @Autowired
    @Qualifier("resultDAO")
    ResultAccess resultDAO;
    @Autowired
    @Qualifier("traceDAO")
    TraceAccess traceDAO;

    @Autowired
    Runs runs;

    public Computation() {
    }

    public Run initialize(Long runId) {
        Run run = runs.getRun(runId);
        resultDAO.purge(run);
        dfs = curveDAO.getDfs(run);
        pds = curveDAO.getPds(run);
        ent = entityDAO.getByRun(run);
        return run;
    }

    public Map<Map.Entry<String, String>, Curve> prepareEpe(Run run, int start, int max) {
        Map<Map.Entry<String, String>, Curve> epe = new HashMap<Map.Entry<String, String>, Curve>();
        epe = dealDAO.getEpe(run, start, max);
        return epe;
    }

    public void calculateResult(Run run, Map<Map.Entry<String, String>, Curve> epes) {
        List<Result> results = new ArrayList<Result>();

        Curve pdSPECIAL = pds.get(SPECIAL);

        for (Entry<Entry<String, String>, Curve> entry : epes.entrySet()) {
            Entity ntt = ent.get(entry.getKey().getValue());
            if (ntt == null) {
                continue;
            }

            double lgd;
            if (ntt.getLgd() == null) {
                traceDAO.setTrace(new Trace(0, entry.getKey().getKey(), ntt.getId(), run.getRunId(), 
                        "Missing data", "Missing lgd, no computation", new Date()));
                results.add(new Result(entry.getKey().getKey(), ntt.getId(), null, run.getRunId()));
                continue;
            } else {
                lgd = ntt.getLgd();
            }

            Curve pd = pds.get(ntt.getId());
            String currency = ntt.getCurrency();
            Curve df;

            if (currency == null || currency.isEmpty()) {
                traceDAO.setTrace(new Trace(0, entry.getKey().getKey(), ntt.getId(), run.getRunId(), "Forcing",
                        "No currency found, USD used", new Date()));
                df = dfs.get("USD");
            } else {
                df = dfs.get(currency);
            }

            if (pd == null) {
                traceDAO.setTrace(new Trace(0, entry.getKey().getKey(), ntt.getId(), run.getRunId(),
                        "Missing data", "Missing pd, no computation", new Date()));
            }
            if (df == null) {
                traceDAO.setTrace(new Trace(0, entry.getKey().getKey(), ntt.getId(), run.getRunId(),
                        "Missing data", "Missing df, no computation", new Date()));
            }
            if (pd == null || df == null) {
                results.add(new Result(entry.getKey().getKey(), ntt.getId(), null, run.getRunId()));
                continue;
            }

            double result = 0.0;
            double probaDownSPECIAL = 0.0;

            for (int i = 0; i < Constants.STEPS; i++) {
                try {
                    double epe = entry.getValue().get(i);
                    double discountFactor = df.get(i);
                    double probaDown = pd.get(i);
                    if (i > 0) {
                        probaDownSPECIAL += pdSPECIAL.get(i - 1);
                    }
                    result += epe * discountFactor * probaDown * (1 - probaDownSPECIAL);
                } catch (NullPointerException npe) {
                    npe.printStackTrace();
                }
            }
            result *= lgd;
            results.add(new Result(entry.getKey().getKey(), ntt.getId(), result, run.getRunId()));
        }
        resultDAO.setResult(results);
    }

    /**
     * getters and setters...
     */

    public static void main(String[] args) {
        Long runId = Long.parseLong(args[0]);

        ApplicationContext context = new ClassPathXmlApplicationContext("classpath*:applicationContext.xml");
        Computation computer = context.getBean(Computation.class);
        Run run = computer.initialize(runId);

        Map<Map.Entry<String, String>, Curve> epe = new HashMap<Map.Entry<String, String>, Curve>();

        int start = 0;

        do {
            epe.clear();
            epe = computer.prepareEpe(run, start, MAX_RESULTS);
            computer.calculateResult(run, epe);
            start += MAX_RESULTS;
        } while (!(epe.size() < MAX_RESULTS));
    }
}

The ResultDAO (where I write the results) :

@Repository("resultDAO")
@Transactional("resultTxManager")
public class ResultDAO extends EntityAccess implements ResultAccess {
    public final static String DELETE_RESULTS = "delete from RESULTS where RUN_ID = ?";

    @Autowired
    @Qualifier("resultSessionFactory")
    private SessionFactory sessionFactory;
    private Session session;

    @Override
    public void setResult(List<Result> results) throws HibernateException {
        if (session == null) {
            session = sessionFactory.openSession();
        }
        Transaction tx = session.beginTransaction();

        int count = 0;
        for (Result row : results) {
            session.save(row);

            if (count++ % 1000 == 0) {
                session.flush();
                session.clear();
            }
        }
        tx.commit();
    }

    @Override
    public void purge(Run run) throws HibernateException {
        if (session == null) {
            session = sessionFactory.openSession();
        }
        Transaction t = session.beginTransaction();
        try {
            SQLQuery query = session.createSQLQuery(DELETE_RESULTS);
            query.setLong(0, run.getRunId());
            query.executeUpdate();
            t.commit();
        } catch (HibernateException he) {
            t.rollback();
            throw he;
        }
    }
}

Does the session flush frequency have any impact on performance ?

The DealDAO (where I load the EPE table) :

@Repository("dealDAO")
@Transactional("resultTxManager")
public class DealDAO extends EntityAccess implements DealAccess {

    public static final String ALL_EPE_QUERY = "select DEAL_ID, NTT_ID, EPE_TS1, EPE_TS2,"
            + " EPE_TS3, EPE_TS4, EPE_TS5, EPE_TS6, EPE_TS7, EPE_TS8, EPE_TS9,"
            + " EPE_TS10, EPE_TS11, EPE_TS12, EPE_TS13, EPE_TS14, EPE_TS15, EPE_TS16,"
            + " EPE_TS17, EPE_TS18, EPE_TS19, EPE_TS20, EPE_TS21, EPE_TS22,"
            + " EPE_TS23, EPE_TS24, EPE_TS25, EPE_TS26, EPE_TS27, EPE_TS28, EPE_TS29,"
            + " EPE_TS30, EPE_TS31, EPE_TS32, EPE_TS33, EPE_TS34, EPE_TS35, EPE_TS36,"
            + " EPE_TS37, EPE_TS38, EPE_TS39, EPE_TS40, EPE_TS41, EPE_TS42, EPE_TS43,"
            + " EPE_TS44, EPE_TS45, EPE_TS46, EPE_TS47, EPE_TS48, EPE_TS49, EPE_TS50,"
            + " EPE_TS51, EPE_TS52, EPE_TS53, EPE_TS54, EPE_TS55, EPE_TS56, EPE_TS57"
            + " from EPE join RUNS on EPE.EPE_CNT_ID = RUNS.EPE_CNT_ID"
            + " where RUN_ID = ? order by DEAL_ID";

    @Autowired
    @Qualifier("resultSessionFactory")
    private SessionFactory sessionFactory;
    private Session session;


    @Override
    public Map<Map.Entry<String, String>, Curve> getEpe(Run run, int start, int max)
            throws HibernateException {
        if (session == null) {
            session = sessionFactory.openSession();
        }
        SQLQuery select = session.createSQLQuery(ALL_EPE_QUERY);
        select.setLong(0, run.getRunId());
        select.setFirstResult(start).setMaxResults(max);
        @SuppressWarnings("unchecked")
        List result = select.list();

        Map<Map.Entry<String, String>, Curve> epes = new HashMap<Map.Entry<String, String>, Curve>();

        for (Object o : result) {
            Object[] objArray = (Object[]) o;
            Map.Entry<String, String> entry = new AbstractMap.SimpleEntry<String, String>((String) objArray[0],
                    (String) objArray[1]);
            List valles = new LinkedList(Arrays.asList(objArray));
            valles.remove(0);
            valles.remove(0);
            Object[] values = valles.toArray();
            Curve curve = new Curve(values);
            epes.put(entry, curve);
        }
        return epes;
    }
}

OK, the getEpe method is a bit weird, because I am re-using a Curve class instead of creating a new one more suited to my needs. I'll rewrite that part.

Still, how do I choose the size of my batch select (here it is set at 50.000, roughly a tenth of the total number of rows) ?

I didn't include the other DAOs, because I think my question is long enough, and nothing fancy happens in these. I can include them upon request.

Answer 1

Naming, naming, naming...
Your code is very hard to read - ntt, epe, lgd, SPECIAL = "207"(?!?). I understand (hope?) that some of these names are part of your business nomenclature, but, especially when asking strangers to read your code for optimization, giving meaningful names can go a long way. Asking me to figure out how to optimize a loop with result += epe * discountFactor * probaDown * (1 - probaDownSPECIAL) would be hard enough if the names there meant something to me...

Break it down
Your methods are very long (calculateResult is 67 lines long...) this is a code smell, which also makes your code hard to read, as well as hides the structure of your flow, and possible complexity issues.
Refactor your methods to be shorter, with expressive names, that will make your code more readable, more testable, and more optimizeable.

Distribute the load
As @Dan suggested in his comments - you should consider spreading the load on several threads/CPUs/machines...
As far as I could see, your calculations are independent of each other, so you could simply arbitrarily split the load on several instances (let's say take over 9 of your friends' machines, and give each machine all the run ids ending with a specific digit...).
This should give you linear scale (up to the point where writing to the DB is your bottleneck).

Optimizations for one-time code
Perhaps I misunderstood your explanation, but it seems that this code is meant to run once. If that is the case, why do you care if it runs 15 minutes, or 8 hours?

Just run it and get on with your life!