postgresql数据定时转存mongodb方案

我就是我 2021-07-26 23:33 426阅读 0赞

*  案例背景

很多事件记录在最初一段时间读写比较频繁，存储在postgresql比较合适，后期数据量变大，且仅作为历史记录查询，更适合存储在mongodb中，可能需要定期将postgresql中的数据转存到mongodb。

*  案例分析

postgresql数据定时转存mongodb，可以采用jdbc方式将postgresql读入内存，对每条数据进行解析，转成json格式，按mongodb的语法写入数据库。这种方法弊端是，需要对各表建立实体对象，依次解析，或者用更不优雅的方式拼接字符串，整个方案扩展性低，可维护性差。因此，我们采用kettle来完成数据的postgresql读取和mongodb插入脚本编写，并在java工程中进行kettle调用。

*  实现方案

**1****、transformation**

采用transformation完成针对数据的基础转换，执行SQL和数据postgresql读取和mongodb插入等，job则设定定时任务调用transformation,完成对整个工作流程的控制。

1、表输入

表输入查询要转存到mongodb中的数据记录，提供数据源。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70][]

其中数据库连接，要编辑数据库连接类型、IP、端口、密码、数据库名等信息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70 1][]

2、json输出

json输出中操作选择Output value,json条目名称输入json名称，输出值也选择Output value，字段中选择需要插入的字段。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70 2][]

3、插入mongodb

在Big Data中选择MongoDB Output插件，在Configure connection中输入连接信息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70 3][]

在Output options中设置插入时的模式

Truncate:插入数据之前先把集合里的所有数据删除，此模式慎用，仅用于首次全量同步；

Update:更新数据，存在则修改，不存在不操作。

upsert:更新添加，找到匹配项则修改，找不到匹配项则添加。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70 4][]

如果设置了Update模式，在Mongo document fields中需要选择某一字段的Modifier operation设为N/A,其他字段设为$set，表示将该字段作为更新标准，当该字段值存在时，其他字段更新。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70 5][]

可以在Create/drop indexes中创建索引。

![20200822111722563.png][]

注：mongodb插件可能遇到报错：java.lang.NoClassDefFoundError: javax/crypto/spec/PBEKeySpec

在data-integration\\system\\karaf\\etc\\config.properties中org.osgi.framework.bootdelegation参数添加“,javax.crypto,javax.crypto.\*”，保证能正常加载加解密相关类。

**2****、job**

在start中选择合适的时间间隔，以图为例10分钟执行一次。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70 6][]

将之前的transformation添加至转换中

![20200822111826831.png][]

**3****、java工程调用**

1、导入jar包

将kettle基础的jar包导入工程，另外将mongodb驱动mongo-java-driver-2.13.0.jar也导入工程中

2、导入插件

将\\data-integration\\system\\karaf\\system\\pentaho 目录下的插件文件夹pentaho-mongo-utils和pentaho-mongodb-plugin导入工程根目录

3、传入kettle job脚本路径，如果有需要传入脚本中的参数，在params中设定入参：

public static void main(String[] args) throws IOException {
    
          String[] params = { "101", "content", "" };
    
          runJob(params,"E:\\\\dict_data\\pgToMongodb.kjb");
    
       }

4、将需要的参数在job中设定，设置日志级别，ERROR错误日志，BASIC基础日志，ROWLEVEL行级日志，环境初始化并运行job：

public static boolean runJob(String\[\] params, String jobPath) \{  
      try \{  
         KettleEnvironment.*init*();  
         // jobPath是Job脚本的路径及名称          
         JobMeta jobMeta = new JobMeta(jobPath, null);  
         Job job = new Job(null, jobMeta);  
         job.setLogLevel(LogLevel.*ROWLEVEL*);  
         job.start();  
         job.waitUntilFinished();  
         // 向Job 脚本传递参数，脚本中获取参数值：$\{参数名\}  
         job.setLogLevel(LogLevel.*ROWLEVEL*);  
         job.start();  
         job.waitUntilFinished();  
         if (job.getErrors() > 0) \{  
            return false;  
         \} else \{  
            return true;  
         \}  
      \} catch (Exception e) \{  
         e.printStackTrace();  
      \}  
      return false;  
   \}

*  结束语

1、 kettle中postgresql数据转存mongodb，单线程任务Truncate模式下每秒6000-11000条数据，Update模式下每秒600-1000条数据，相差10倍效率，初次全量同步时尽量采用Truncate模式。

2、要控制合理的同步时间间隔，避免过于频繁导致数据传输失败。

3、数据转存不能做到实时，没有数据库连接守护进程，只能重连一定次数，如果数据库连接可能断开，需要另外的补偿机制。

4、本文中完成了在同一表中查询，可以输入postgresql关系型数据库中联表查询的结果，保证适当的冗余，建立合适的索引。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70]: /images/20210726/455231d31ebf4a0ebfb0e86674cab139.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70 1]: /images/20210726/6a680d1794cf45a3bf6d088bb687cbc4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70 2]: /images/20210726/8cea5e6d88c64332994d9f5fa2d417a3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70 3]: /images/20210726/7bc7768039ba4481be286b5427d580c1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70 4]: /images/20210726/583d99695d034a88922e7db898ec5143.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70 5]: /images/20210726/bf5f77cea89b42768c7d208ce83fd953.png
[20200822111722563.png]: /images/20210726/ac39af09dc2444d2bf6c70c0bc0ec126.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1YW5zag_size_16_color_FFFFFF_t_70 6]: /images/20210726/8fe147123dc14b62b0988311bc017d34.png
[20200822111826831.png]: /images/20210726/1f351a71c06d47ef8e1db166b95e63df.png