将Oracle中的数据放入elasticsearch

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();
    }
}

原文地址：https://www.cnblogs.com/JoeyWong/p/9186263.html

时间： 2024-10-28 03:14:33

将Oracle中的数据放入elasticsearch的相关文章

ARM子函数定义中的参数放入寄存器的规则

关于ARM子函数定义中的参数放入寄存器的规则,网上也有很多文章和书籍介绍了,但是还有很多同学不太了解这个基础知识,这里摘取网上的一些介绍,简述一下. 对于ARM体系来说,不同语言撰写的函数之间相互调用(mix calls)遵循的是 ATPCS(ARM-Thumb Procedure Call Standard),ATPCS主要是定义了函数呼叫时参数的传递规则以及如何从函数返回,关于ATPCS的详细内容可以查看ADS1.2 Online Books Developer Guide的2.1节. 简单

几百万数据放入内存不会把系统撑爆吗？

在公司有一个需求是要核对一批数据,之前的做法是直接用SQL各种复杂操作给怼出来的,不仅时间慢,而且后期也不好维护,就算原作者来了过一个月估计也忘了SQL什么意思了,于是有一次我就想着问一下之前做这个需求的人为什么不将这些数据查出来后在内存里面做筛选呢?直接说了你不怕把内存给撑爆吗?此核算服务器是单独的服务器,配置是四核八G的,配置堆的大小是4G.本着怀疑的精神,就想要弄清楚几百万条数据真的放入内存的话会占用多少内存呢? 计算机的存储单位计算机的存储单位常用的有bit.Byte.KB.MB.GB.

MyBatis在Oracle中插入数据并返回主键的问题解决

引言:  在MyBatis中,希望在Oracle中插入数据之时,同时返回主键值,而非插入的条数... 环境:MyBatis 3.2 , Oracle, Spring 3.2   SQL Snippet in XML Configuration: <insert id="insertSelective" parameterType="com.jxxx.p2pp.model.UUserInfo"> <selectKey resultType="

oracle中的数据对象

oracle中的数据对象有表.视图.索引.序列等 表的相关操作 1.创建表 方式一: 方式二:create table person( create table person1 id number(18), as name varchar2(5), select * from person age number(3), sex varchar2(4) ); 2.删除表 方式一:只会删除表中的内容,不会删除表结构truncate delete 方式二:删除表结构truncate table per

Sqoop_具体总结 使用Sqoop将HDFS/Hive/HBase与MySQL/Oracle中的数据相互导入、导出

一.使用Sqoop将MySQL中的数据导入到HDFS/Hive/HBase 二.使用Sqoop将HDFS/Hive/HBase中的数据导出到MySQL 2.3 HBase中的数据导出到mysql 眼下没有直接的命令将HBase中的数据导出到MySQL.但能够先将HBase中的数据导出到HDFS中.再将数据导出到MySQL. 三.使用Sqoop将Oracle中的数据导入到HDFS/Hive/HBase 以下仅仅给出将Oracle中的数据导入HBase,其它情况下的命令行选项与MySQL的操作相似

详细总结 使用Sqoop将HDFS/Hive/HBase与MySQL/Oracle中的数据相互导入、导出

一.使用Sqoop将MySQL中的数据导入到HDFS/Hive/HBase 二.使用Sqoop将HDFS/Hive/HBase中的数据导出到MySQL 2.3 HBase中的数据导出到mysql 目前没有直接的命令将HBase中的数据导出到MySQL,但可以先将HBase中的数据导出到HDFS中,再将数据导出到MySQL. 三.使用Sqoop将Oracle中的数据导入到HDFS/Hive/HBase 下面只给出将Oracle中的数据导入HBase,其他情况下的命令行选项与MySQL的操作相似 O

在Oracle中更新数据时，抛出：ORA-01008： not all variables bound

在Oracle中更新数据时,抛出了一个 :ORA-01008 not all variables bound, 我的理解是不是所有的变量/参数都有边界,不懂: 后来知道了,原来是“不是所有变量/参数都确定”, 就是有些变量没有指定,缺少变量参数, 最后发现是因为在写三层时少写了一个"new OracleParameter(":ID",userinfo.ID);" 导致的.

Oracle中读取数据一些原理研究

文章很多摘录了 http://blog.163.com/[email protected]/blog/static/7956964020131069843572/ 同时基于这篇文章的基础上,补充一些学习要点,如有问题,希望指出探讨. 1 ORACLE体系结构 下图描述了oracle的体系结构.SGA(system global area)是各个进程共享的内存块,Buffer cache用来缓存数据文件的数据块(block). 2 如何在data buffer中查找数据块 data buffer存

运用html5+css3+jq+js实现添加的数据放入本地存储和界面

<!DOCTYPE html> <html> <head lang="en"> <meta charset="UTF-8"> <title>历史记录</title> <style> *{ margin: 0; padding: 0; } body{ margin-left: 300px; } ul{ list-style: none; } ul li,div{ width: 250p