Применение JDBC в процессе ETL

В предыдущей статье мы упоминали о Java + JDBC в качестве альтернативы Python + ODBC. В этой статье приведём условную ситуацию, приближенную к той, которая побудила нас сменить язык программирования.

Была поставлена задача по обработке более 25 млн. строк посредством процесса ETL: на основе имеющихся данных расcчитать два дополнительных столбца и вернуть их в базу. Python прекрасно справлялся с подобными задачами, но в данной ситуации скорость получения результата была не самой оптимальной: обработка длилась более суток.

Так как процедуру предполагалось выполнять многократно, то было решено найти более быстрый инструмент. Поиск альтернативных библиотек для Python ни к чему не привёл. Поэтому было решено применить тяжёлую артиллерию — Java, так как Java работает с БД быстрее, чем Python чисто технически. И чем больше объём данных, тем заметнее реализуется это преимущество.

На первом этапе разовьём класс DBC из предыдущей статьи, добавив:

1. метод execQuerySelect для выполнения select-запросов в разных вариантах (с TOP, where и т.д.)

2. метод execQueryInsert для вставки результата в БД

3. вспомогательные метода buildInsertColumns и buildInsertValues для формирования колонок и строк соответственно.

В итоге класс DBC пополнится следующим кодом:

/**
     * Выполнение select запроса к БД
     *
     * @param select Названия необходимых колонок (* - получить все)
     * @param from   Таблица из которой небходимо получить данные
     * @return ResultSet с результатами работы запроса
     */
    public ResultSet execQuerySelect(String select, String from) {
        if (!readyToWork) {
            System.err.println("DBC not ready to work! Abort:execQuerySelected");
            return null;
        }

        String query = "SELECT " + select + " FROM " + from;

        return execQuery(query);
    }

    /**
     * Выполнение select запроса к БД
     *
     * @param select Названия необходимых колонок (* - получить все)
     * @param from   Таблица из которой небходимо получить данные
     * @param top    Количество "верхних" строк получаемых из базы
     * @return ResultSet с результатами работы запроса
     */
    public ResultSet execQuerySelect(String select, String from, Integer top) {
        if (!readyToWork) {
            System.err.println("DBC not ready to work! Abort:execQuerySelected");
            return null;
        }

        String query = "SELECT TOP(" + top + ") " + select + " FROM " + from;

        return execQuery(query);
    }

    /**
     * Выполнение select запроса к БД
     *
     * @param select Названия необходимых колонок (* - получить все)
     * @param from   Таблица из которой небходимо получить данные
     * @param where  Условие выборки значений из таблицы
     * @return ResultSet с результатами работы запроса
     */
    public ResultSet execQuerySelect(String select, String from, String where) {
        if (!readyToWork) {
            System.err.println("DBC not ready to work! Abort:execQuerySelected");
            return null;
        }

        String query = "SELECT " + select + " FROM " + from + " WHERE " + where;

        return execQuery(query);
    }

    /**
     * Выполнение select запроса к БД
     *
     * @param select Названия необходимых колонок (* - получить все)
     * @param from   Таблица из которой небходимо получить данные
     * @param where  Условие выборки значений из таблицы
     * @param top    Количество "верхних" строк получаемых из базы
     * @return ResultSet с результатами работы запроса
     */
    public ResultSet execQuerySelect(String select, String from, String where, Integer top) {
        if (!readyToWork) {
            System.err.println("DBC not ready to work! Abort:execQuerySelected");
            return null;
        }

        String query = "SELECT TOP(" + top + ") " + select + " FROM " + from + " WHERE " + where;

        return execQuery(query);
    }

    /**
     * Выполнение insert запроса к БД
     *
     * @param table Имя таблицы для вставки
     * @param columns Массив колонок для вставки
     * @param data Вектор словарей со значениями для вставки
     * @return Флаг успешности выполнения запроса
     */
    public boolean execQueryInsert(String table, String[] columns, Vector<HashMap<String, String>> data) {
        if (!readyToWork) {
            System.err.println("DBC not ready to work! Abort:execQuerySelected");
            return false;
        }

        String query = "INSERT INTO " + table + " " + buildInsertColumns(columns) + " VALUES ";

        for (int i = 0; i < data.size() - 1; i++) {
            query = query.concat(buildInsertValues(columns, data.get(i), false));
        }

        query = query.concat(buildInsertValues(columns, data.get(data.size() - 1), true));

        execQuery(query);
        return true;
    }
	
	    /**
     * Формирует строку с перечнем колонок для запроса insert
     *
     * @param columns Массив колонок
     * @return Строку с перечнем колонок для запроса insert
     */
    private String buildInsertColumns (String[] columns) {

        String r = "(";

        for (int i = 0; i < columns.length - 1; i++) {
            r = r.concat("[" + columns[i] + "], ");
        }

        r = r.concat("[" + columns[columns.length-1] + "])");

        return r;
    }

    /**
     * Формирует строку с перечнем значений для запроса insert
     *
     * @param columns Массив колонок
     * @param data Словарь значений для вставки
     * @param isLast Флаг нахождения значений на последней позиции среди остальных
     * @return Строку с перечнем значений для запроса insert
     */
    private String buildInsertValues (String[] columns, HashMap<String, String> data, boolean isLast) {
        String vals = "(";

        for (int i = 0; i < columns.length - 1; i++) {
            vals = vals.concat(data.get(columns[i]).concat(", "));
        }

        vals = vals.concat(data.get(columns[columns.length-1]).concat(")"));

        if (isLast) {
            vals = vals.concat(";");
        }
        else
        {
            vals = vals.concat(", ");
        }

        return vals;
    }

Теперь, имея все необходимые инструменты, перейдём к решению нашей задачи. В рамках абстракции заменим сложные вычисления на вычисления суммы и разности двух чисел. Итоговый алгоритм заключается в получении двух чисел из одной таблицы, проведении расчетов, вставке результата в новую таблицу. С учётом алгоритма решения нашей задачи MainClass из предыдущей статьи будет выглядеть так:

package DataBaseTools;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Vector;

public class MainClass {

    public static void main(String[] args) throws SQLException {
        // create instance
        DBC dbc = new DBC("localhost", "testDB");


        // get query response
        ResultSet response = dbc.execQuerySelect("x, y", "digits",);


        Vector<HashMap<String, String>> insertBuffer = new Vector<HashMap<String, String>>();

        // handle received data
        if (response == null) {
            System.out.println("NULL");
        } else {
            while (response.next()) {
                HashMap<String, String> rowData = new HashMap<String, String>();

				int x = response.getInt("x");
				int y = response.getInt("y");
				
				int a = x + y;
				int s = x - y;
				
				rowData.put("a", a.toString());
				rowData.put("s", s.toString());

                insertBuffer.add(rowData);

                if (insertBuffer.size() == 1000) {
                    String[] cols = {"a", "s"};

                    System.out.println(dbc.execQueryInsert("[results]", cols, insertBuffer));

                    insertBuffer.clear();
                }
            }

            if (insertBuffer.size() > 0) {
                String[] cols = {"a", "s"};

                System.out.println(dbc.execQueryInsert("[results]", cols, insertBuffer));

                insertBuffer.clear();
            }
        }

    }
}

Этот код создаёт подключение к базе, получает данные из исходной таблицы, производит расчет и запись результата в буфер, а при наполнении буфера происходит запись в базу. После выполнения цикла оставшиеся в буфере данные также записываются в базу. Обработка такого количества данных при изначальной сложности вычислений заняла чуть более 7 часов, что минимум в 4 раза быстрее аналогичного кода на Python. При увеличении скорости работы с БД такое решение позволит увеличить сложность вычислений, не увеличивая время выполнения.