自己写的数据交换工具——从Oracle到Elasticsearch

本文涉及的产品
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
简介:

先说说需求的背景,由于业务数据都在Oracle数据库中,想要对它进行数据的分析会非常非常慢,用传统的数据仓库-->数据集市这种方式,集市层表会非常大,查询的时候如果再做一些group的操作,一个访问需要一分钟甚至更久才能响应。

为了解决这个问题,就想把业务库的数据迁移到Elasticsearch中,然后针对es再去做聚合查询。

问题来了,数据库中的数据量很大,如何导入到ES中呢?

Logstash JDBC

Logstash提供了一款JDBC的插件,可以在里面写sql语句,自动查询然后导入到ES中。这种方式比较简单,需要注意的就是需要用户自己下载jdbc的驱动jar包。

input {
    jdbc {
        jdbc_driver_library => "ojdbc14-10.2.0.3.0.jar"
        jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
        jdbc_connection_string => "jdbc:oracle:thin:@localhost:1521:test"
        jdbc_user => "test"
        jdbc_password => "test123"
        schedule => "* * * * *"
        statement => "select * from TARGET_TABLE"
        add_field => ["type","a"]
    }
}
output{
    elasticsearch {
        hosts =>["10.10.1.205:9200"]
        index => "product"
        document_type => "%{type}"
    }
}

不过,它的性能实在是太差了!我导了一天,才导了两百多万的数据。

因此,就考虑自己来导。

自己的数据交换工具

思路:

449064-20160923181130934-1775007112.png

最后使用发现,自己写的导入程序,比Logstash jdbc快5-6倍~~~~~~ 嗨皮!!!!

遇到的问题

  • 1 JDBC需要采用分页的方式读取全量数据
  • 2 要模仿bulk文件进行存储
  • 3 由于bulk文件过大,导致curl内存溢出

程序开源

下面的代码需要注意的就是

public class JDBCUtil {
    private static Connection conn = null;
    private static PreparedStatement sta=null;
    static{
        try {
            Class.forName("oracle.jdbc.driver.OracleDriver");
            conn = DriverManager.getConnection("jdbc:oracle:thin:@localhost:1521:test", "test", "test123");
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        }
        System.out.println("Database connection established");
    }
    /**
    * 把查到的数据格式化写入到文件
    *
    * @param list 需要存储的数据
    * @param index 索引的名称
    * @param type 类型的名称
    * @param path 文件存储的路径
    **/
    public static void writeTable(List<Map> list,String index,String type,String path) throws SQLException, IOException {
        System.out.println("开始写文件");
        File file = new File(path);
        int count = 0;
        int size = list.size();
        for(Map map : list){
            FileUtils.write(file,  "{ \"index\" : { \"_index\" : \""+index+"\", \"_type\" : \""+type+"\" } }\n","UTF-8",true);
            FileUtils.write(file, JSON.toJSONString(map)+"\n","UTF-8",true);
//            System.out.println("写入了" + ((count++)+1) + "[" + size + "]");
        }
        System.out.println("写入完成");
    }

    /**
     * 读取数据
     * @param sql
     * @return
     * @throws SQLException
     */
    public static List<Map> readTable(String tablename,int start,int end) throws SQLException {
        System.out.println("开始读数据库");
        //执行查询
        sta = conn.prepareStatement("select * from(select rownum as rn,t.* from "+tablename+" t )where rn >="+start+" and rn <"+end);
        ResultSet rs = sta.executeQuery();

        //获取数据列表
        List<Map> data = new ArrayList();
        List<String> columnLabels = getColumnLabels(rs);

        Map<String, Object> map = null;
        while(rs.next()){
            map = new HashMap<String, Object>();

            for (String columnLabel : columnLabels) {
                Object value = rs.getObject(columnLabel);
                map.put(columnLabel.toLowerCase(), value);
            }
            data.add(map);
        }
        sta.close();
        System.out.println("数据读取完毕");
        return data;
    }
    /**
     * 获得列名
     * @param resultSet
     * @return
     * @throws SQLException
     */
    private static List<String> getColumnLabels(ResultSet resultSet)
            throws SQLException {
        List<String> labels = new ArrayList<String>();

        ResultSetMetaData rsmd = (ResultSetMetaData) resultSet.getMetaData();
        for (int i = 0; i < rsmd.getColumnCount(); i++) {
            labels.add(rsmd.getColumnLabel(i + 1));
        }

        return labels;
    }
    /**
    * 获得数据库表的总数,方便进行分页
    *
    * @param tablename 表名
    */
    public static int count(String tablename) throws SQLException {
        int count = 0;
        Statement stmt = conn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_UPDATABLE);
        ResultSet rs = stmt.executeQuery("select count(1) from "+tablename);
        while (rs.next()) {
            count = rs.getInt(1);
        }
        System.out.println("Total Size = " + count);
        rs.close();
        stmt.close();
        return count;
    }
    /**
     * 执行查询,并持久化文件
     * 
     * @param tablename 导出的表明
     * @param page 分页的大小
     * @param path 文件的路径
     * @param index 索引的名称
     * @param type 类型的名称
     * @return
     * @throws SQLException
     */
    public static void readDataByPage(String tablename,int page,String path,String index,String type) throws SQLException, IOException {
        int count = count(tablename);
        int i =0;
        for(i =0;i<count;){
            List<Map> map = JDBCUtil.readTable(tablename,i,i+page);
            JDBCUtil.writeTable(map,index,type,path);
            i+=page;
        }
    }
}

在main方法中传入必要的参数即可:

public class Main {
    public static void main(String[] args) {
        try {
            JDBCUtil.readDataByPage("TABLE_NAME",1000,"D://data.json","index","type");
        } catch (SQLException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

这样得到bulk的数据后,就可以运行脚本分批导入了。

下面脚本的思路,就是每100000行左右的数据导入到一个目标文件,使用bulk命令导入到es中。注意一个细节就是不能随意的切分文件,因为bulk的文件是两行为一条数据的。

#!/bin/bash

count=0
rm target.json
touch target.json


while read line;do

((count++))

{
        echo $line >> target.json

        if [ $count -gt 100000 ] && [ $((count%2)) -eq 0 ];then
                count=0
                curl -XPOST localhost:9200/_bulk --data-binary @target.json > /dev/null
                rm target.json
                touch target.json
        fi

}

done < $1
echo 'last submit'
curl -XPOST localhost:9200/_bulk --data-binary @target.json > /dev/null

最后执行脚本:

sh auto_bulk.sh data.json

自己测试最后要比logstasj jdbc快5-6倍。

本文转自博客园xingoo的博客,原文链接:自己写的数据交换工具——从Oracle到Elasticsearch,如需转载请自行联系原博主。
相关实践学习
使用阿里云Elasticsearch体验信息检索加速
通过创建登录阿里云Elasticsearch集群,使用DataWorks将MySQL数据同步至Elasticsearch,体验多条件检索效果,简单展示数据同步和信息检索加速的过程和操作。
ElasticSearch 入门精讲
ElasticSearch是一个开源的、基于Lucene的、分布式、高扩展、高实时的搜索与数据分析引擎。根据DB-Engines的排名显示,Elasticsearch是最受欢迎的企业搜索引擎,其次是Apache Solr(也是基于Lucene)。 ElasticSearch的实现原理主要分为以下几个步骤: 用户将数据提交到Elastic Search 数据库中 通过分词控制器去将对应的语句分词,将其权重和分词结果一并存入数据 当用户搜索数据时候,再根据权重将结果排名、打分 将返回结果呈现给用户 Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索,具有接近实时的搜索,并支持多租户。
相关文章
|
1月前
|
关系型数据库 MySQL
elasticsearch对比mysql以及使用工具同步mysql数据全量增量
elasticsearch对比mysql以及使用工具同步mysql数据全量增量
29 0
|
1月前
|
监控 Java 测试技术
【Elasticsearch专栏 13】深入探索:Elasticsearch使用Curator工具删除Elasticsearch中的历史数据
使用Curator工具可以有效管理Elasticsearch中的旧数据,通过编写YAML配置文件定义删除操作。配置中指定了基于索引名称前缀和年龄的过滤器,确保仅删除符合条件的旧索引。执行删除操作时,Curator会应用过滤器识别目标索引,并向Elasticsearch发送删除请求。通过设置选项,如忽略空列表和超时时间,可以确保操作的灵活性和稳定性。使用Curator不仅释放了存储空间,还提高了查询性能,是维护Elasticsearch健康的重要工具
57 0
|
1天前
|
运维 监控 数据可视化
cerebro【ElasticSearch可视化工具】
cerebro【ElasticSearch可视化工具】
8 1
|
6天前
|
SQL Oracle 关系型数据库
探索 Linux 命令 `db_archive`:Oracle 数据库归档日志的工具
探索 Linux 中的 `db_archive`,实际与 Oracle 数据库归档日志管理相关。在 Oracle 中,归档日志用于恢复,当在线重做日志满时自动归档。管理员可使用 SQL*Plus 查看归档模式,通过 `RMAN` 进行备份和恢复操作。管理归档日志需谨慎,避免数据丢失。了解归档管理对 Oracle 管理员至关重要,确保故障时能快速恢复数据库。
|
8天前
|
存储 安全 数据管理
如何在 Rocky Linux 8 上安装和配置 Elasticsearch,帮助你快速搭建起这个强大的工具
【6月更文挑战第7天】本文档详细介绍了如何在Rocky Linux 8上安装和配置Elasticsearch,首先通过添加Elasticsearch仓库并使用yum安装。接着,配置Elasticsearch,包括修改`elasticsearch.yml`、设置内存和文件描述符,以及可选的安全设置。启动Elasticsearch后,通过验证日志和测试连接确保其正常运行。文章还列举了常见问题及解决方法,如启动失败、内存不足和网络问题。按照这些步骤,用户可以在Rocky Linux 8上成功部署Elasticsearch,为数据管理与分析提供强大支持。
26 5
|
7月前
|
JSON 数据格式 索引
实际使用Elasticdump工具对Elasticsearch集群进行数据备份和数据还原
就可以通过Elasticsearch的导入导出工具Elasticdump来实现,可以将Elasticsearch不同集群的数据进行索引备份和还原。
124 0
|
7月前
|
存储 数据可视化 搜索推荐
分布式系列教程(26) -分布式日志搜集工具Elasticsearch简介
分布式系列教程(26) -分布式日志搜集工具Elasticsearch简介
65 0
|
1月前
|
Oracle 关系型数据库
Oracle 管理诊断数据工具ADRCI
Oracle 管理诊断数据工具ADRCI
19 2
|
1月前
|
自然语言处理 数据可视化 算法
史上最详细Docker安装Elasticsearch、ik分词器、可视化工具,每一步都带有步骤图!!!
史上最详细Docker安装Elasticsearch、ik分词器、可视化工具,每一步都带有步骤图!!!
541 0
史上最详细Docker安装Elasticsearch、ik分词器、可视化工具,每一步都带有步骤图!!!
|
8月前
|
Oracle 关系型数据库 数据库连接
Windows系统安装配置Oracle数据库连接工具PLSQL
Windows系统安装配置Oracle数据库连接工具PLSQL