// This is what boots Cassandra for the hadoop stuff 
private void startKeyspace() throws IOException
{
    CalloutManager.instance().onStart();
    LogUtil.init();
    
    if (keyspace == null) {
        throw new RuntimeException("you forgot to set the keyspace with KeyspaceInputFormat.setKeyspace()");
    }
    
    keyspace = DatabaseDescriptor.getTable(keyspace);
    if (keyspace == null) {
        throw new RuntimeException("that keyspace doesn't exist in cassandra!");
    }
    
    Set<String> tables = DatabaseDescriptor.getTableToColumnFamilyMap().keySet();
    for (String table : tables) 
    {
        Table tbl = Table.open(table);
        tbl.onStart();
    }
    
    RecoveryManager recoveryMgr = RecoveryManager.instance();
    recoveryMgr.doRecovery();
 
    StorageService.instance().start(); // NPE is here.
}