数据同步
TIP
不同数据库之间数据同步
全量同步(oracle-mogodb)
//源数据SQL
const sqlOracle = `select GOODID as "product_id",
goods as "product_code",
GNAME as "product_name",
spec as "product_spec",
MSUNITNO as "product_unit",
PRODUCER as "producer_name",
PRDNO as "producer_code",
WAREBRAND as "warebrand"
from PUB_WAREDICT
`
//创建MongoDB低代码数据库的连接事物
const instanceOrg = await api.database.getMongoConnection('JFMbPYSBTDKdjBsd');
//获取目标数据
const old = await instanceOrg.collection('DnlqvoR012t6Utvr').find().toArray();
//定义目标数据主键
const oldMap = {}
for (const u of old) {
oldMap[u.product_code] = u
}
//定义新增、更新数组
const insertlist = []
const updatelist = []
//获取当前时间
let datatime = api.utils.formatDate(new Date(), 'YYYY-MM-DD HH:mm:ss');
try {
//查询源数据条数
const sqlCountsql = `select count(*) as count from PUB_WAREDICT`
const sqlCount = await api.database.executeOracleQuery('GnstHtXSZWNmtnnT', sqlCountsql);
const QueryCount = sqlCount.rows[0].COUNT;
//循环判断数据新增/更新
if (QueryCount > 0) {
const sqlresult = await api.database.executeOracleQuery('GnstHtXSZWNmtnnT', sqlOracle);
for (const v of sqlresult.rows) {
const a = v.product_code
const or = oldMap[a]
if (or) {
updatelist.push({
updateOne: {
filter: { _id: or._id },
update: {
$set: {
updateCode: 'admin',
updateName: '系统管理员',
updateTime: datatime,
product_name: v.product_name,
brand_name: v.warebrand,
product_spec: v.product_spec,
product_unit: v.product_unit,
producer_code: v.producer_code,
producer_name: v.producer_name,
}
}
}
})
}
else {
insertlist.push({
_id: a,
createName: '系统管理员',
createCode: 'admin',
createTime: datatime,
product_code: v.product_code,
product_name: v.product_name,
brand_name: v.warebrand,
product_spec: v.product_spec,
product_unit: v.product_unit,
producer_code: v.producer_code,
producer_name: v.producer_name,
})
}
}
//批量执行更新
if (updatelist.length !== 0) {
await instanceOrg.collection('DnlqvoR012t6Utvr').bulkWrite(updatelist)
}
//批量执行新增
if (insertlist.length !== 0) {
await instanceOrg.collection('DnlqvoR012t6Utvr').insertMany(insertlist)
}
}
}
catch (e) {
api.logger.info(e)
//异常情况回滚
// instanceDir.rollback();
}
finally {
//执行完成后关闭连接,mogodb数据库不需要关闭连接
// await instanceOrg.close();
}
全量同步(mogodb-postgres)
//获取当前时间
let curdatetime = api.utils.formatDate(new Date(), 'YYYY-MM-DD HH:mm:ss');
//创建MongoDB低代码数据库的连接事物
const instanceOrg = await api.database.getMongoConnection('JFMbPYSBTDKdjBsd');
//创建Postgre数仓测试数据库的连接事物
const instanceDir = await api.database.getPostgresInstance('RfhbbHQpnQRnXWnG');
try {
//获取源数据
_resSqltext = await instanceOrg.collection('jTMWYdADPHsECyfp').find({ "user_code": { "$ne": null, $exists: true } }).toArray();
//查询源数据条数
api.logger.info('总记录：[', _resSqltext.length, ']')
//映射外部json对象键名称，正常情况肯定不一致，都需要映射
const newData_order = _resSqltext.map(v => ({
owner_code: v.owner_code,
owner_name: v.owner_name,
customer_code: v.customer_code,
customer_name: v.customer_name,
seller_code: v.seller_code,
seller_name: v.seller_name,
user_code: v.user_code,
user_name: v.user_name,
operating_time: curdatetime,
pkey: v.owner_code.concat(v.department_code, v.customer_code)
}))
if (newData_order) {
//定义表名
const tableName = 'dw_multilevel_push_relationship';
//定义INSERT 栏位名
const colsInsert =
[
'owner_code',
'owner_name',
'customer_code',
'customer_name',
'seller_code',
'seller_name',
'user_code',
'user_name',
'operating_time',
'pkey'
];
//定义 UPDATE 栏位名
const colsUpdate = ['owner_code',
'owner_name',
'customer_code',
'customer_name',
'seller_code',
'seller_name',
'user_code',
'user_name',
'operating_time'];
//定义 更新条件
const onCond = ['pkey']
//分步执行新增、更新操作
const res = await api.sql.postgresUpdateConflict(tableName, colsInsert, colsUpdate, onCond, newData_order);
await instanceDir.executeBatch(res[0], res[1]);
}
//统一提交
instanceDir.commit();
}
catch (e) {
api.logger.info(e)
//异常情况数据回滚
instanceDir.rollback();
}
finally {
//执行完成后关闭连接,mogodb数据库不需要关闭连接
await instanceDir.close();
}
清空插入同步(mogodb-postgres)
//创建MongoDB低代码数据库的连接事物
const mogodb = await api.database.getMongoConnection('rGTGJyPPPhtDDAnT')
let sqlCount = 0
let insertlist = []
//定义插入字段
const colums = [
'id',
'code',
'name',
'spec',
'unit',
"producer",
'status',
'approvedocNo'
]
try {
//查找源数据数量
sqlCount = await mogodb.collection('standard_goods').countDocuments()
if (sqlCount > 0) {
//获取源数据
const userList = await mogodb.collection('standard_goods').find().limit(10).toArray()
insertlist = userList.map(u => ({
id: u.id,
code: u.code,
name: u.name,
spec: u.spec,
unit: u.unit,
producer: u.producer,
status: u.status,
approvedocNo: u.approvedocNo
})
)
//清空目标表
await api.database.executePostgresQuery('wSECzZxKNEyamkRE', 'truncate dc_standard_goods')
//执行插入
const insertSql_oprat = api.sql.postgresInsertManyText('dc_standard_goods', colums, insertlist)
await api.database.executePostgresQuery('wSECzZxKNEyamkRE', insertSql_oprat[0], insertSql_oprat[1]);
}
}
catch (e) {
api.logger.info(e)
}
finally {
}