Kettle增量导出MongoDB到Mysql表中

一、需求：

        将MongoDB表中的数据按照时间戳增量抽取到Mysql表中。

二、实现方式：

      1. kettle 

      2. python 脚本

三、遇到的问题：

         kettle 如何将增量时间作为变量传入到MongoDB Input 中？

         翻遍了百度，没有一篇文章能写明白的，本人将实现方式详尽记载下来，供大家参考！

四、具体实现：

        1. kettle实现

           实现思路：

                    step1： 先全量将MongoDB 导入到 Mysql 表中。（这个不难，百度中有实现。）

                    step2：创建第一个Transformation，获取Mysql 中最大时间字段 例如：Max(time),将其存入到Mysql的一张表中。

                        step2-1: 配置表输入（目的：为了取得全量表中时间字段最大值）

                        step2-2：设置变量

                        step2-3：将变量值输出到表T1中

                    step3：创建第二个Transformation，配置MongoDB Input -> Mysql

                            step3-1 表输入

                            step3-2 设置变量

                            step3-3 配置MongoDB Input

                            step3-4 配置字段选择

                            step3-5 表输出 

                    step4：创建作业执行step2 和 step3 两个转换。

            具体步骤如下：

            step1：先全量将MongoDB 导入到 Mysql 表中。（略）

mongo_test 表（全量表）

step1-1

            step2：创建第一个Transformation，获取Mysql 中最大时间字段 例如：Max(time),将其存入到Mysql的一张表中。

step2
step2-1 表输入
step2-2 设置变量
step2-3 将变量值保存到表t1中

step2 t1 表

step3 ：创建第二个Transformation，配置MongoDB Input -> Mysql

step3

 step3-1 表输入

step3-2 设置变量

step3-3 配置MongoDB Input

注意写法

这里需要注意：STRAT_TIME是step3-2中的变量名，这样就可以获取到变量值

step3-4 配置字段选择

step3-5 表输出

step4：创建作业执行step2 和 step3 两个转换。

测试：

1. 当前t1 中的时间为：

        2018-05-19 00:00:00    

2. 向MongoDB中添加一条数据

> db.xk.insert({&＃8220;name&＃8221;:&＃8221;发发发发发财&＃8221;,&＃8221;age&＃8221;:&＃8221;30&＃8243;,&＃8221;time&＃8221;:&＃8221;2018-05-20 00:00:00&＃8243;})

3. 执行job

4. 查看结果：

以上结束：

python 脚本方式也很简单：

==============================================================

#coding=utf-8

# Python 实现 从MongoDB 想MySql 增量导数据

from pymongo import MongoClient

import pymysql

# mongodb

client = MongoClient(&＃8216;192.168.107.128&＃8217;, 27017)

TempleSpider = client[&＃8216;xk&＃8217;]

temple_comment_collect = TempleSpider[&＃8216;xk&＃8217;]

# mysql

mysql = pymysql.connect(&＃8216;localhost&＃8217;, &＃8216;root&＃8217;, &＃8216;123456&＃8217;, &＃8216;test&＃8217;, charset=&＃8221;utf8&＃8243;)

# insert sql

sql = &＃8216;insert into mongo_test values(%s,%s,%s)&＃8217;

# 批量提交

def batch_commit(connect, target_sql, result):

    cursor = connect.cursor()

    list = []

    i = 0

    try:

        for row in result:

            data = (row[&＃8216;name&＃8217;].encode(&＃8216;utf-8&＃8217;), row[&＃8216;age&＃8217;], row[&＃8216;time&＃8217;])

            print(row[&＃8216;name&＃8217;])

            list.append(data)

            if i >= 1000:

                cursor.executemany(target_sql, list)

                connect.commit()

                print(&＃8216;1000 条插入成功&＃8230;&＃8217;)

                i = 0

                list.clear()

            i = i+1

        if i > 0:

            cursor.executemany(target_sql, list)

            connect.commit()

            print(&＃8216;%d 条插入成功&＃8230;&＃8230;&＃8217; % i)

    except Exception as e:

        print(e)

        print(&＃8220;增量数据导入失败！！！&＃8221;)

        # 回滚

        connect.rollback()

records = temple_comment_collect.find({&＃8220;time&＃8221;: {&＃8220;$gt&＃8221;: &＃8220;2018-05-15 00:00:00&＃8221;}})

batch_commit(mysql, sql, records)

client.close()

mysql.close()

==============================================================