本文最后更新于 676 天前,其中的信息可能已经有所发展或是发生改变。
对价格字段进行增量同步。
简单阐述全量同步和增量同步区别:
全量同步:就是每天定时(避开业务高峰期)或者周期性全量把数据从一个地方拷贝到另外一个地方,可以采用直接全部覆盖(使用新数据覆盖旧数据)或者走更新逻辑(覆盖前判断下,如果新旧不一致,就更新)。
增量同步:就是指抓取某个时刻(更新时间)或者检查点以后的数据来同步,不是无规律的全量同步。
1、基础的配置文件
MongoDB 写插件的官方文档:Datax MongoDBWriter
{
"job": {
"setting": {
"speed": {
"channel": 3
},
"errorLimit": {}
},
"content": [
{
"reader": {
"name": "postgresqlreader",
"parameter": {
"username": "{from_username}",
"password": "{from_password}",
"where": "",
"column": [
"{from_column_01}",
"{from_column_02}"
],
"connection": [
{
"table": [
"{from_table}"
],
"jdbcUrl": [
"jdbc:postgresql://{from_host}:{from_port}/{from_database}"
]
}
]
}
},
"writer": {
"name": "mongodbwriter",
"parameter": {
"address": [
"{to_host}:{to_port}"
],
"userName": "{to_username}",
"userPassword": "{to_password}",
"dbName": "{to_database}",
"collectionName": "smr_{web}_{mode}_history",
"column": [
{
"name": "{from_column_01}",
"type": "string"
},
{
"name": "{from_column_02}",
"type": "string"
}
]
}
}
}
]
}
}
2、我的需求和解决
我的需求是:定时将 PostgreSQL 表中的价格字段和更新时间字段,以更新时间的时间戳作为主键,保存到 MongoDB 中。
写起来还是很简单的,只要注意一下 PostgreSQL 时间戳的转换就行了。
最后的配置文件:
{
"job": {
"setting": {
"speed": {
"channel": 10
},
"errorLimit": {
"record": 2,
"percentage": 0.02
}
},
"content": [
{
"reader": {
"name": "postgresqlreader",
"parameter": {
"username": "{from_username}",
"password": "{from_password}",
"where": "",
"column": [
"update_time",
"price"
],
"connection": [
{
"table": [
"test_table"
],
"jdbcUrl": [
"jdbc:postgresql://{from_host}:{from_port}/{from_database}"
]
}
]
}
},
"writer": {
"name": "mongodbwriter",
"parameter": {
"address": [
"{to_host}:{to_port}"
],
"userName": "{to_username}",
"userPassword": "{to_password}",
"dbName": "{to_database}",
"collectionName": "test_table_price_history",
"column": [
{
"name": "time",
"type": "int"
},
{
"name": "price",
"type": "double"
}
],
"upsertInfo": {
"isUpsert": "false"
}
}
}
}
]
}
}
结束。