使用datax将postgresql或者greenplum中的数据同步到elasticsearch
使用datax将postgresql或者greenplum中的数据同步到elasticsearch
1、使用datax工具将postgresql或者greenplum数据库中的数据同步到elasticsearch中。DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、NOSQL、大数据计算系统都已经接入，目前支持数据如下图：
类型 | 数据源 | Reader(读) | Writer(写) | 文档 |
|---|---|---|---|---|
RDBMS 关系型数据库 | MySQL | √ | √ | 读 、写 |
Oracle | √ | √ | 读 、写 | |
SQLServer | √ | √ | 读 、写 | |
PostgreSQL | √ | √ | 读 、写 | |
DRDS | √ | √ | 读 、写 | |
通用RDBMS(支持所有关系型数据库) | √ | √ | 读 、写 | |
阿里云数仓数据存储 | ODPS | √ | √ | 读 、写 |
ADS | √ | 写 | ||
OSS | √ | √ | 读 、写 | |
OCS | √ | √ | 读 、写 | |
NoSQL数据存储 | OTS | √ | √ | 读 、写 |
Hbase0.94 | √ | √ | 读 、写 | |
Hbase1.1 | √ | √ | 读 、写 | |
Phoenix4.x | √ | √ | 读 、写 | |
Phoenix5.x | √ | √ | 读 、写 | |
MongoDB | √ | √ | 读 、写 | |
Hive | √ | √ | 读 、写 | |
Cassandra | √ | √ | 读 、写 | |
无结构化数据存储 | TxtFile | √ | √ | 读 、写 |
FTP | √ | √ | 读 、写 | |
HDFS | √ | √ | 读 、写 | |
Elasticsearch | √ | 写 | ||
时间序列数据库 | OpenTSDB | √ | 读 | |
TSDB | √ | √ | 读 、写 |
可以看到Elasticsearch只支持写，但是不支持读的，如果支持从Elasticsearch读出来，写到postgresql或者greenplum也是很好的哦！
2、datax的安装，配置就不写了，之前搞过，现在需要搞一个postgresql或者greenplum写到elasticsearch的json，需要注意的是需要安装一个postgresqlreader读插件的，将插件放到datax\datax\plugin\reader中。
然后，需要搞一个elasticsearchwriter写插件，将elasticsearchwriter插件放在datax\datax\plugin\writer中。
1 {
2 "job": {
3 "setting": {
4 "speed": {
5 "byte": 8388608,
6 "channel": 3
7 },
8 "errorLimit": {
9 "record": 0,
10 "percentage": 0.02
11 }
12 },
13 "content": [{
14 "reader": {
15 "name": "postgresqlreader",
16 "parameter": {
17 "username": "数据库账号",
18 "password": "数据库密码",
19 "connection": [
20 {
21 "querySql": [
22 "SELECT "update_time",\"UserName\",\"Age\" from core_data.up_lic_data where \"Id\" > 1 and \"ID\" < 10000 limit 1000 ;"
23 ],
24 "jdbcUrl": ["jdbc:postgresql://数据库ip地址:数据库端口号/数据库名称"]
25 }
26 ]
27 }
28 },
29 "writer": {
30 "name": "elasticsearchwriter",
31 "parameter": {
32 "endpoint": "http://es主节点ip地址:es端口号",
33 "index": "user_info", // 索引名称
34 "type": "doc", // 文档类型
35 "cleanup": false,
36 "dynamic": false,
37 "settings": {
38 "index": {
39 "number_of_shards": 5, // 分片数量
40 "number_of_replicas": 1 // 副本数量
41 }
42 },
43 "batchSize": 10000,
44 "splitter": ",",
45 "column": [
46 {
47 "name": "update_time",
48 "type": "keyword"
49 },{
50 "name": "UserName",
51 "type": "keyword"
52 },{
53 "name": "Age",
54 "type": "keyword"
55 }
56 ]
57 }
58 }
59 }]
60 }
61 }