Shulou(Shulou.com)06/01 Report--

This article mainly explains "how to import elasticsearch into mysql data", the content of the explanation is simple and clear, easy to learn and understand, the following please follow the editor's ideas slowly in depth, together to study and learn "how to achieve elasticsearch import mysql data" bar!

I. The official batch import of API based on elasticsearch introduces maven dependency org.springframework.boot spring-boot-starter-parent 2.3.2.RELEASE org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-data-elasticsearch Mysql mysql-connector-java runtime

Jdbc connection class

Public class DBHelper {public static final String url = "jdbc:mysql://localhost:3306/lagou_db?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai"; public static final String name = "com.mysql.cj.jdbc.Driver"; public static final String user = "root"; public static final String password = "root"; public static Connection conn = null; public static Connection getConn () {try {Class.forName (name) Conn = DriverManager.getConnection (url, user, password); / / get connection} catch (Exception e) {e.printStackTrace ();} return conn;}} Import Logic @ Service ("positionService") public class PositionService {@ Autowired ElasticsearchRestTemplate elasticsearchTemplate; @ Autowired RestHighLevelClient client; private static final String POSITIOIN_INDEX = "position"; public void importAll () throws IOException {writeMysqlDataToES (POSITIOIN_INDEX) } / * * write data in bulk to ES * / private void writeMysqlDataToES (String tableName) {BulkProcessor bulkProcessor = getBulkProcessor (client); Connection conn = null; PreparedStatement ps = null; ResultSet rs = null; try {conn = DBHelper.getConn (); System.out.println ("Start handle data:" + tableName); String sql = "SELECT * from" + tableName Ps = conn.prepareStatement (sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); / / set ps.setFetchSize (20) according to your needs; rs = ps.executeQuery (); ResultSetMetaData colData = rs.getMetaData (); ArrayList dataList = new ArrayList () / / there are not many ways for bulkProcessor to add data support. Check its api and find that it supports map key-value pairs, so the author converts the checked data into hashMap mode HashMap map = null; int count = 0; String c = null; String v = null; while (rs.next ()) {count++ Map = new HashMap; for (int I = 1; I client .bulkAsync (request, RequestOptions.DEFAULT, bulkListener); BulkProcessor.Builder builder = BulkProcessor.builder (bulkConsumer, listener); builder.setBulkActions (5000); builder.setBulkSize (new ByteSizeValue (100L, ByteSizeUnit.MB)) Builder.setConcurrentRequests (10); builder.setFlushInterval (TimeValue.timeValueSeconds (100L)); builder.setBackoffPolicy (BackoffPolicy.constantBackoff (TimeValue.timeValueSeconds (1L), 3)); / / Note: let the parameter settings take effect bulkProcessor = builder.build ();} catch (Exception e) {e.printStackTrace () Try {bulkProcessor.awaitClose (100L, TimeUnit.SECONDS);} catch (Exception E1) {System.out.println (e1.getMessage ());}} return bulkProcessor;}} call entry @ RestControllerpublic class PositionController {@ Autowired PositionService positionService @ RequestMapping ("query") public List query (String positionName) {if (positionName = = null) {return null;} return positionService.queryPositions (positionName);} @ RequestMapping ("/ importAll") public String importAll () {try {positionService.importAll ();} catch (IOException e) {e.printStackTrace () } return "success";}} imported data table public class Position implements Serializable {/ / key private String id; / / company name private String companyName; / / position name private String positionName; / / position temptation private String positionAdvantage; / / salary private String salary; / / salary floor private int salaryMin; / / salary ceiling private int salaryMax / / private String education; / / working years private String workYear; / / release time private String publishTime; / / work city private String city; / / work place private String workAddress; / / release time @ JsonFormat (pattern = "yyyy-MM-dd HH:mm:ss") private Date createTime; / / work mode private String jobNature;} II. Import based on logstash

Prerequisite: install logstash

Import.confinput {stdin {} jdbc {jdbc_connection_string = > "jdbc:mysql://localhost:3306/lagou_db?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai" jdbc_user = > "root" jdbc_password = > "root" jdbc_driver_library = > "D:/mysql-connector-java-5.1.10.jar" jdbc_driver_class = > "com.mysql.jdbc.Driver" jdbc_ Paging_enabled = > "true" jdbc_page_size = > "1000" statement_filepath = > "D:/import.sql"}} filter {json {source = > "message" remove_field = > ["message"]}} output {elasticsearch {hosts = > ["localhost:9200"] index = > "position" document_type = > "_ doc"} stdout {codec = > json_lines}} import.sqlselect * from position launch logstashlogstash-f.. / import.conf Thank you for your reading The above is the content of "how to import elasticsearch into mysql data". After the study of this article, I believe you have a deeper understanding of how to import elasticsearch into mysql data, and the specific use needs to be verified in practice. Here is, the editor will push for you more related knowledge points of the article, welcome to follow!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.