自己写的数据交换工具——从Oracle到Elasticsearch
时间:2022-04-22
本文章向大家介绍自己写的数据交换工具——从Oracle到Elasticsearch,主要内容包括Logstash JDBC、自己的数据交换工具、程序开源、基本概念、基础应用、原理机制和需要注意的事项等,并结合实例形式分析了其使用技巧,希望通过本文能帮助到大家理解应用这部分内容。
先说说需求的背景,由于业务数据都在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}"
}
}
不过,它的性能实在是太差了!我导了一天,才导了两百多万的数据。
因此,就考虑自己来导。
自己的数据交换工具
思路:
- 1 采用JDBC的方式,通过分页读取数据库的全部数据。
- 2 数据库读取的数据存储成bulk形式的数据,关于bulk需要的文件格式,可以参考这里
- 3 利用bulk命令分批导入到es中
最后使用发现,自己写的导入程序,比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倍。
- 进程监控工具supervisor 启动Mongodb
- 祭奠那些年,我弃坑的开源轮子
- 这些奇技浮巧,助你优化前端应用性能
- Stepping.js——两步完成前后端分离架构设计
- 我的职业是前端工程师【十】客户端存储艺术:数据存储与模型
- 【开源】2md:将复制的内容、网页转成 markdown
- React Native 持续部署实践— push 代码构建出新版的 Growth
- 技巧 - 如何好一个 Git 提交信息及几种不同的规范
- React、Vue、Ember 及其他前端开发者,请暂缓更新到 Chrome 59 浏览器
- 微软开源全新的文档生成工具DocFX
- 使用 MimeKit 和 MailKit 发送邮件
- 使用 React Native 重写大型 Ionic 应用后,我们想分享一下这八个经验
- 基于OWin的Web服务器Katana发布版本3
- 【工具推荐】图像界的魔术师 ImageMagick
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法