將Oracle中的數據放入elasticsearch
阿新 • • 發佈：2018-06-15
tid public div ont message pid req getdate sql
package com.c4c.test; import java.sql.Connection; import java.sql.DriverManager; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; import java.util.ArrayList; import java.util.Date; import java.util.List; public class OracleToETS { static final String JDBC_DRIVER = "oracle.jdbc.driver.OracleDriver";//"oracle.jdbc.driver.OracleDriver";//驅動 static final String DB_URL = "jdbc:oracle:thin:@127.0.0.1:1521:s2";//數據庫URL static final String USER ="joey";//數據庫對應的用戶名 static final String PASS = "joey";//登錄密碼 public static void main(String[] args) { OracleToETS tmp = new OracleToETS(); List<User> userList = tmp.getAllList(); ESClient esclient = new ESClient(); esclient.initESClient(); esclient.bulkdoc(userList); } public List<User> getAllList() { ResultSet rs = null;//執行SQL語句，返回的結果 Statement stmt = null; //將數據發送到數據庫中 Connection conn = null;//建立數據庫的連接 List<User> list = new ArrayList<User>(); try { Class.forName(JDBC_DRIVER);//oracle數據庫驅動 conn = DriverManager.getConnection(DB_URL,USER,PASS);//數據庫url 用戶名 密碼 stmt = conn.createStatement(); rs = stmt.executeQuery("SELECT * FROM o_user "); while(rs.next()) { String tmpId = rs.getString("ID"); String tmpName = rs.getString("NAME"); String tmpAddress = rs.getString("ADDRESS"); String tmpetype = rs.getString("E_TYPE"); String tmpeindustry = rs.getString("E_INDUSTRY"); Date tmpcheckdate = rs.getDate("CHECK_DATE"); list.add(new User(tmpId,tmpName,tmpAddress,tmpetype,tmpeindustry,rs.getString("EMPIRICAL_SCOPE"),tmpcheckdate, rs.getDate("CREATE_TIME"),rs.getString("LEGAL_PERSON"),rs.getString("REGISTRATION_UNIT"),rs.getString("JURISDICTION_UNIT"), rs.getString("PHONE_NO"),rs.getString("BAR_ID"),rs.getString("BUILD_ID"),rs.getString("FLOOR_ID"), rs.getString("ROOM_ID"),rs.getString("BAR_LNG"),rs.getString("BAR_LAT"),rs.getString("BUILD_LNG"),rs.getString("BUILD_LAT"), rs.getString("ROOM_LNG"),rs.getString("ROOM_LAT"))); } }catch(Exception e) { e.printStackTrace(); }finally { try { conn.close(); } catch (SQLException e) { e.printStackTrace(); } } return list; } }
package com.c4c.test; import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; import java.text.SimpleDateFormat; import java.util.Date; import java.util.List; import java.util.Map; import java.util.Set; import org.apache.lucene.analysis.compound.hyphenation.TernaryTree.Iterator; import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse; import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.action.delete.DeleteResponse; import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.search.SearchRequestBuilder; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.search.SearchType; import org.elasticsearch.client.Client; import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.transport.InetSocketTransportAddress; import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.xcontent.XContentFactory; import org.elasticsearch.index.query.QueryBuilder; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.aggregations.Aggregation; import org.elasticsearch.search.aggregations.AggregationBuilders; import org.elasticsearch.search.aggregations.bucket.filters.Filters.Bucket; import org.elasticsearch.search.aggregations.bucket.terms.StringTerms; import org.elasticsearch.search.aggregations.bucket.terms.TermsBuilder; import org.elasticsearch.search.sort.FieldSortBuilder; import org.elasticsearch.search.sort.SortOrder; import org.junit.After; import org.junit.Before; /** * @author joey * */ public class ESClient { private Client client; /** * 初始化客戶端連接 */ @Before public Client initESClient() { // 配置你的es,如果你的集群名稱不是默認的elasticsearch，需要以下這步 Settings settings = Settings.settingsBuilder().put("cluster.name","elasticsearch").build(); // 這裏可以同時連接集群的服務器,可以多個,並且連接服務是可訪問的 try { // on startup，因為是默認的集群名，所以不需要setting client = TransportClient.builder().build() .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("127.0.0.1"), 9300)); } catch (UnknownHostException e) { e.printStackTrace(); }finally { if(client != null) { System.out.println("連接成功！"); } } return client; } @After public void closeESClient() { client.close(); System.out.println("連接關閉！"); } /** * 創建索引、類型、文檔並插入ID為1文檔內容 */ @SuppressWarnings("unused") private void createIndex() { IndexResponse response = null; try { response = client.prepareIndex("customertest", "joey", "1") .setSource(XContentFactory.jsonBuilder().startObject() .field("name", "joey") .field("time", new Date()) .endObject() ) .get(); } catch (IOException e) { e.printStackTrace(); }finally { if(response != null) { System.out.println("索引創建成功！"); } } } /** * 獲取文檔 */ public void get(){ GetResponse response = client.prepareGet("customer", "external", "1").get(); Set<String> headers = response.getHeaders(); System.out.println(headers);//獲取請求頭 boolean exists = response.isExists(); System.out.println(exists);//判斷索引是否存在 String sourceString = response.getSourceAsString(); System.out.println(sourceString);//獲取索引，並打印出索引內容 String id = response.getId(); System.out.println(id);//獲取索引id boolean sourceEmpty = response.isSourceEmpty(); System.out.println(sourceEmpty);//獲取索引的內容是否為空 } /** * 刪除 */ public void delete(){ //DeleteResponse response = client.prepareDelete("customertest", "joey", "1").get(); //刪除文檔 //刪除索引 DeleteIndexResponse deleteIndexResponse = client.admin().indices() .prepareDelete("s2") .execute().actionGet(); boolean isFound = deleteIndexResponse.isAcknowledged(); System.out.println(isFound);//返回文檔是否存在，存在刪除 } /** * 搜索索引 */ public void search(){ //創建查詢索引 SearchRequestBuilder searchRequestBuilder = client.prepareSearch("s2"); //設置查詢索引類型 searchRequestBuilder.setTypes("o_user"); //設置查詢類型 searchRequestBuilder.setSearchType(SearchType.DEFAULT); //設置查詢關鍵字 searchRequestBuilder.setQuery(QueryBuilders.commonTermsQuery("name", "寧波")); // 設置查詢數據的位置,分頁用吧 searchRequestBuilder.setFrom(0); // 設置查詢結果集的最大條數 searchRequestBuilder.setSize(10); // 設置是否按查詢匹配度排序 searchRequestBuilder.setExplain(true); // 最後就是返回搜索響應信息 SearchResponse response = searchRequestBuilder.execute().actionGet(); //System.out.println("響應結果："+response); //獲取搜索文檔的結果 SearchHits searchHits = response.getHits(); SearchHit[] hits = searchHits.getHits(); System.out.println("數量："+hits.length); for (int i = 0; i < hits.length; i++) { SearchHit hit = hits[i]; Map<String, Object> result = hit.getSource(); System.out.println("文檔結果："+result); } System.out.println("查詢索引完畢!"); } /** * 批量新增 */ public void bulkdoc(List<User> list) { SimpleDateFormat dateFormater = new SimpleDateFormat("yyyy-MM-dd"); BulkRequestBuilder bulkRequest = client.prepareBulk(); long b = System.currentTimeMillis(); try { for(User user:list) { String _id = user.getId(); bulkRequest.add(client.prepareIndex("s2", "o_user", _id) .setSource(XContentFactory.jsonBuilder() .startObject() .field("name", user.getName()) .field("address",user.getAddress()) .field("etype",user.getE_type()) .field("e_industry",user.getE_industry()) .field("empirical_scope",user.getEmpirical_scope()) .field("checkdate",dateFormater.format(user.getCheck_date())) .field("create_time",dateFormater.format(user.getCreate_time())) .field("legal_person",user.getLegal_person()) .field("registration_unit",user.getRegistration_unit()) .field("jurisdiction_unit",user.getJurisdiction_unit()) .field("phone_no",user.getPhone_no()) .field("bar_id",user.getBar_id()) .field("build_id",user.getBuild_id()) .field("floor_id",user.getFloor_id()) .field("room_id",user.getRoom_id()) .field("bar_lng",user.getBar_lng()) .field("bar_lat",user.getBar_lat()) .field("build_lng",user.getBuild_lng()) .field("build_lat",user.getBuild_lng()) .field("room_lng",user.getRoom_lng()) .field("room_lat",user.getRoom_lat()) .endObject() ) ); } } catch (IOException e) { e.printStackTrace(); } BulkResponse bulkResponse = bulkRequest.execute().actionGet(); if (bulkResponse.hasFailures()) { System.out.println(bulkResponse.buildFailureMessage()); } long useTime = System.currentTimeMillis()-b; System.out.println("useTime:" + useTime); } /** * 聚合查詢 */ public void aggSearch() { SearchRequestBuilder srb = client.prepareSearch("s2"); srb.setSize(0); srb.setTypes("o_user"); srb.setHighlighterType("o_user"); srb.setSearchType(SearchType.QUERY_THEN_FETCH); TermsBuilder typeTermsBuilder = AggregationBuilders.terms("etypeAgg").field("etype"); typeTermsBuilder.size(100); srb.addAggregation(typeTermsBuilder); SearchResponse sr = srb.execute().actionGet(); System.out.println(sr); } public void useScrolls() { QueryBuilder qb = QueryBuilders.multiMatchQuery("工廠", "name"); SearchResponse scrollResp = client.prepareSearch("s2") .addSort("name", SortOrder.ASC) .setScroll(new TimeValue(60000)) .setQuery(qb) .setSize(100).execute().actionGet(); //100 hits per shard will be returned for each scroll while (true) { for (SearchHit hit : scrollResp.getHits().getHits()) { System.out.println(hit.getScore()); } scrollResp = client.prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(60000)).execute().actionGet(); if (scrollResp.getHits().getHits().length == 0) { break; } } } public static void main(String args[]){ System.out.println("ES集群連接測試"); ESClient esc = new ESClient(); esc.initESClient(); //esc.createIndex(); //esc.get(); //esc.delete(); //esc.search(); //esc.bulkdoc(); // esc.closeESClient(); esc.aggSearch(); //esc.useScrolls(); } }
將Oracle中的數據放入elasticsearch