目录

前言

 编写代码

报错解决

完整代码 

---

前言

        一直以来的都是使用TableOutput输出到表，但是如果是有主键，需要更新数据，用这个很明显是不可以的。

        所以需要InsertUpdateMeta节点，交换模型如下图红框所示，表输入-->插入/更新

         插入更新节点的设置如下图

两个表的话如下图所示，

目标表无主键

 

 源表有主键

 

运行一次上述交换后，源表数据插入到目标表，再次运行发现目标表还是只有两条数据，如下图所示，

         这个和tableoutput不一样的就是，tableoutput每次运行都会把源表数据插入一份到目标表。如果tableoutput运行了两次，那么表里面应该有4条数据了。

        修改源表数据，运行后发现目标表多了一条数据，并没有更新，这是因为目标表没有主键，无法比对，

         给目标表加上主键，再运行一次

 

         如下图所示，运行后更新了数据，而不是新插入一条，这是和tableoutput不一样的，插入更新节点可以根据主键更新数据。

 编写代码

        为了下面字段区分，我将目标表的字段名改了。

 插入更新对应代码对象

InsertUpdateMeta insertUpdateMeta = new InsertUpdateMeta();

设置查询关键字

//设置查询关键字
String[] destPk = new String[]{"id_d"};
String[] srcPk = new String[]{"id"};

insertUpdateMeta.setKeyLookup(destPk);
insertUpdateMeta.setKeyStream(srcPk);//流里的字段对应输入

insertUpdateMeta.setKeyStream2(destPk);//这个代码是报错后加的

上面代码对应下图红框处设置，

 搞不懂为何代码中要设置keyStream2，而spoon工具中明明不需要设置流里的字段2就可以交换。

设置关键字出的查询条件，

String[] condition = new String[]{"="};
insertUpdateMeta.setKeyCondition(condition);//对比条件

设置更新字段

//设置更新的字段
String[] destFields = new String[]{"id_d","name_d"};
String[] srcFields = new String[]{"id","name"};

insertUpdateMeta.setUpdateStream(srcFields);
insertUpdateMeta.setUpdateLookup(destFields);

 

设置更新字段处的是否更新标志，

//设置是否更新
Boolean[] updateFlags = new Boolean[]{true,true};
insertUpdateMeta.setUpdate(updateFlags);

 最后完整的更新节点代码如下，

StepMeta tableInputStep = new StepMeta(tableInputPluginId,
        "tableInput", (StepMetaInterface) tableInputMeta);
transMeta.addStep(tableInputStep);


InsertUpdateMeta insertUpdateMeta = new InsertUpdateMeta();
insertUpdateMeta.setDatabaseMeta(destDatabaseMeta);

//设置目标表的 schema和表名
insertUpdateMeta.setSchemaName(null);
insertUpdateMeta.setTableName("user_info_dest");

//设置查询关键字
String[] destPk = new String[]{"id_d"};
String[] srcPk = new String[]{"id"};
insertUpdateMeta.setKeyLookup(destPk);
insertUpdateMeta.setKeyStream(srcPk);//流里的字段对应输入
insertUpdateMeta.setKeyStream2(destPk);

String[] condition = new String[]{"="};
insertUpdateMeta.setKeyCondition(condition);//对比条件


//设置更新的字段
String[] destFields = new String[]{"id_d","name_d"};
String[] srcFields = new String[]{"id","name"};
insertUpdateMeta.setUpdateStream(srcFields);
insertUpdateMeta.setUpdateLookup(destFields);

//设置是否更新
Boolean[] updateFlags = new Boolean[]{true,true};
insertUpdateMeta.setUpdate(updateFlags);

String insertUpdatePluginId = registry.getPluginId(StepPluginType.class, insertUpdateMeta);
StepMeta insertUpdateStep = new StepMeta(insertUpdatePluginId, "insertUpdate" , (StepMetaInterface) insertUpdateMeta);

//将步骤添加进去
transMeta.addStep(insertUpdateStep);

//将步骤和上一步关联起来
transMeta.addTransHop(new TransHopMeta(tableInputStep, insertUpdateStep));

报错解决

        运行一下后报错，如下图所示，

         跟踪进代码发现需要设置keyStream2，

         设置keyStream2,如下代码

insertUpdateMeta.setKeyStream2(destPk);

        然后交换正常了，能插入，能根据主键更新。

完整代码 

/**
     * mysql到mysql之间的交换
     * @throws KettleException
     */
    @Test
    public void exchangeMySQL2MySQL() throws KettleException{
        //源数据库连接
        String src = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>" +
                "<connection>" +
                "<name>src</name>" +
                "<server>192.168.10.64</server>" +
                "<type>MYSQL</type>" +
                "<access>Native</access>" +
                "<database>test</database>" +
                "<port>3306</port>" +
                "<username>root</username>" +
                "<password>root</password>" +
                "</connection>";

        //目标数据库连接
        String dest = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>" +
                "<connection>" +
                "<name>dest</name>" +
                "<server>192.168.10.64</server>" +
                "<type>MYSQL</type>" +
                "<access>Native</access>" +
                "<database>test</database>" +
                "<port>3306</port>" +
                "<username>root</username>" +
                "<password>root</password>" +
                "</connection>";

        DatabaseMeta srcDatabaseMeta = new DatabaseMeta(src);
        DatabaseMeta destDatabaseMeta = new DatabaseMeta(dest);

        //创建转换元信息
        TransMeta transMeta = new TransMeta();
        transMeta.setName("mysql到mysql之间的交换");

        //设置源和目标
        transMeta.addDatabase(srcDatabaseMeta);
        transMeta.addDatabase(destDatabaseMeta);

		/*
		 * 创建  表输入->插入/更新
		 * 同时将两个步骤连接起来
		 */
        PluginRegistry registry = PluginRegistry.getInstance();
        TableInputMeta tableInputMeta = new TableInputMeta();
        String tableInputPluginId = registry.getPluginId(StepPluginType.class,
                tableInputMeta);

        tableInputMeta.setDatabaseMeta(srcDatabaseMeta);
        //设置查询条件
        String selectSql = "select id ,name from user_info_src";
        tableInputMeta.setSQL(selectSql);

        StepMeta tableInputStep = new StepMeta(tableInputPluginId,
                "tableInput", (StepMetaInterface) tableInputMeta);
        transMeta.addStep(tableInputStep);


        InsertUpdateMeta insertUpdateMeta = new InsertUpdateMeta();
        insertUpdateMeta.setDatabaseMeta(destDatabaseMeta);

        //设置目标表的 schema和表名
        insertUpdateMeta.setSchemaName(null);
        insertUpdateMeta.setTableName("user_info_dest");

        //设置查询关键字
        String[] destPk = new String[]{"id_d"};
        String[] srcPk = new String[]{"id"};
        insertUpdateMeta.setKeyLookup(destPk);
        insertUpdateMeta.setKeyStream(srcPk);//流里的字段对应输入
        insertUpdateMeta.setKeyStream2(destPk);

        String[] condition = new String[]{"="};
        insertUpdateMeta.setKeyCondition(condition);//对比条件


        //设置更新的字段
        String[] destFields = new String[]{"id_d","name_d"};
        String[] srcFields = new String[]{"id","name"};
        insertUpdateMeta.setUpdateStream(srcFields);
        insertUpdateMeta.setUpdateLookup(destFields);

        //设置是否更新
        Boolean[] updateFlags = new Boolean[]{true,true};
        insertUpdateMeta.setUpdate(updateFlags);

        String insertUpdatePluginId = registry.getPluginId(StepPluginType.class, insertUpdateMeta);
        StepMeta insertUpdateStep = new StepMeta(insertUpdatePluginId, "insertUpdate" , (StepMetaInterface) insertUpdateMeta);

        //将步骤添加进去
        transMeta.addStep(insertUpdateStep);

        //将步骤和上一步关联起来
        transMeta.addTransHop(new TransHopMeta(tableInputStep, insertUpdateStep));

        Trans trans = new Trans(transMeta);

        //执行转换
        trans.execute(null);

        //等待完成
        trans.waitUntilFinished();
        if (trans.getErrors() > 0) {
            System.out.println("交换出错.");
            return;
        }
    }