使用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中。

{
    "job": {
        "setting": {
            "speed": {
                "byte": 8388608,
                "channel": 3
            },
            "errorLimit": {
                "record": 0,
                "percentage": 0.02
            }
        },
        "content": [{
            "reader": {
                "name": "postgresqlreader",
                "parameter": {
                    "username": "數據庫賬號",
                    "password": "數據庫密碼",
                    "connection": [
                        {
                            "querySql": [
                                "SELECT "update_time",\"UserName\",\"Age\" from core_data.up_lic_data where \"Id\" > 1 and \"ID\" < 10000 limit 1000  ;"
                            ],
                            "jdbcUrl": ["jdbc:postgresql://數據庫ip地址:數據庫端口號/數據庫名稱"]
                        }
                    ]
                }
            },
            "writer": {
                "name": "elasticsearchwriter",
                "parameter": {
                    "endpoint": "http://es主節點ip地址:es端口號",
                    "index": "user_info", // 索引名稱
                    "type": "doc", // 文檔類型
                    "cleanup": false, 
                    "dynamic": false,
                    "settings": {
                        "index": {
                            "number_of_shards": 5, // 分片數量
                            "number_of_replicas": 1 // 副本數量
                        }
                    },
                    "batchSize": 10000,
                    "splitter": ",",
                    "column": [
                        {
                            "name": "update_time",
                            "type": "keyword"
                        },{
                            "name": "UserName",
                            "type": "keyword"
                        },{
                            "name": "Age",
                            "type": "keyword"
                        }
                    ]
                }
            }
        }]
    }
}