Плагин Logstash-JDBC: импорт миллионов записей из Oracle в ElasticSearch

Сценарий. – Я пытаюсь импортировать большой набор данных из базы данных Oracle (несколько миллионов записей) в ElasticSearch. - Я могу импортировать данные с помощью плагина ввода logstash jdbc.

Проблема: – Выполнение занимает очень много времени (5 часов).

  • Есть ли лучшее решение, чтобы сократить это время? Хорошая практика в таких случаях?

performance logstash logstash-jdbc elasticsearch
person shan    schedule 27.06.2016    source источник

Ответы (3)


arrow_upward
4
arrow_downward

Сначала вы можете поиграть с jdbc_fetch_size, следующим шагом будет запуск импорта на нескольких узлах, для каждого узла разделите запрос sql на непересекающиеся наборы, такие как select * from XX where id between 0 and 1000000.

Вы также можете установить количество реплик индекса ES равным 0 перед индексированием большой партии и изменить его обратно после. И, конечно же, используйте API массового индексирования с хорошим размером пакета и параллелизмом.

person xeye    schedule 27.06.2016
comment
xeye спасибо за ответ! Чтобы увеличить fetch_size, мне нужно иметь достаточно места в куче для хранения записи выборки. Мне нужно найти лучший fetch_size, который не даст мне исключение OutOfMemory. Можем ли мы использовать какие-либо очереди между DB и ES? как Кафка? - person shan; 27.06.2016
comment
конечно, вы можете использовать kafka, но сама по себе kafka не улучшит производительность индексирования. - person xeye; 28.06.2016
comment
Я планировал кафку, чтобы не удерживать соединение с БД на все время индексации. Как только данные передаются в kafka, нам не нужно подключение к БД. Как увеличить производительность индексации? Я знаю, что мы можем установить flush_size для ElasticSearch, чтобы использовать массовое API. - person shan; 28.06.2016

arrow_upward
2
arrow_downward

Ты можешь использовать

:sql_last_value

возможность получать данные из базы данных постепенно с помощью планировщика. Что-то вроде следующего будет полезно. (Предполагая, что у вас есть поле идентификатора)

input {
   jdbc {
      jdbc_connection_string => "jdbc:oracle:thin:IP:PORT:SID"
      jdbc_user => "user"
      jdbc_password => "password"
      jdbc_driver_library => ".......\ojdbc8.jar"
      jdbc_driver_class => "Java::oracle.jdbc.OracleDriver"
      statement => "select * from (select * from TABLE where id >= :sql_last_value ORDER BY id ASC) rownum < 20"
      use_column_value => true
      tracking_column => id
      tracking_column_type => "numeric"
      schedule => "* * * * *"
  }

}

обновление: рефакторинг SQL для использования rownum и ORDER BY для ограничения результатов в предполагаемом порядке (сортировка перед ограничением). См.: о ROWNUM и ограничении результатов

Тот же результат может быть достигнут с помощью разбиения на страницы, но это имеет проблемы с производительностью. Если мы используем нумерацию страниц ex:

input {
   jdbc {
      jdbc_connection_string => "jdbc:oracle:thin:IP:PORT:SID"
      jdbc_user => "user"
      jdbc_password => "password"
      jdbc_driver_library => ".......\ojdbc8.jar"
      jdbc_driver_class => "Java::oracle.jdbc.OracleDriver"
      jdbc_page_size => 20
      jdbc_paging_enabled => true
      statement => "select * from TABLE"
  } 
}

Но это завернет

"выбрать * из ТАБЛИЦЫ"

нравится

SELECT * FROM (SELECT "T1".*, ROWNUM "X_SEQUEL_ROW_NUMBER_X" FROM (SELECT * FROM (select * from TABLE) "T1") "T1") "T1" WHERE (("X_SEQUEL_ROW_NUMBER_X" > 0) AND ("X_SEQUEL_ROW_NUMBER_X " ‹= (0 + 20)));

и будет работать без планировщика, разделив запрос на основе указанного размера jdbc_page_size (20 для этого примера). Но этот метод, очевидно, имеет проблемы с производительностью.

person Erangad    schedule 19.10.2017

arrow_upward
-1
arrow_downward

Вы можете поместить в один и тот же файл конфигурации больше, чем входной плагин jdbc. У меня он отлично работает.

i.e:

ввод { jdbc { ... }

   jdbc {
       ...
      }

}

person Basem Mohammed    schedule 13.02.2017