借鉴sqoop实现hdfs文件内容导入mysql

这次需要将hadoop mr的计算结果导入到mysql中，虽然是mr的结果导入db中，为了保险起见，还是存在hdfs上，之后读取hdfs上的结果导入db中，读取失败可重新执行单个读取导入过程。

一般先动手前，有个思路，再百度看看是否有更好的实现，大略搜了一下，发现sqoop貌似实现了hdfs和各种dc之间的读取写入。这里，因为业务简单，都是insert语句不涉及事务，只是连接一个db，不涉及mr等操作，so我只是借鉴sqoop的思想，没有使用sqoop。

一般实现的思路就是，读取hdfs文件，生成对应的insert语句，导入mysql就好了。

其中需要详细考虑的几个问题如下：

1、批量导入insert，一般的数据量设置多大好些？

2、执行一般失败后重新导入数据，对于已经导入的数据如何处理？

这里的话，当然批量导入会好些，但是也要考虑hdfs的reader是一个个读取数据，如果批量导入的size太大，需要存储数据的变量占用的内存大，会导致oom。

一般批量insert的量不超过1000条就好了，我这边的话，每天的量也就2000条，so我设置成500条了。

执行一般失败后重新导入数据，感觉要先删除已有导入的数据，再次导入。因为我这边每天导入一次，根据日期做delete就行了。对于不同场景，需要自己操作了。

如果是mysql数据库的话，不用自己删除了。

因为有个语句   INSERT INTO .. ON DUPLICATE KEY UPDATE  ，如果执行时不存在重复记录，则执行新增操作，否则执行更新操作。详细的内容可以读http://www.jb51.net/article/39255.htm

如果db是mysql的话，就不用自己先删除数据了。

另 INSERT INTO TABLE (a,c) VALUES (1,3),(1,7) ON DUPLICATE KEY UPDATE c=VALUES(c);

上语句比单独执行2条语句快。

本人才疏学浅，有错误之处请不吝指教。

关于insert的数量，很有必要深入研究下。

至此说完。。。

------------------------------

贴些代码（部分是sqoop中的源码改编）

    /**
     * 如果记录重复，则执行更新操作，否则执行新增数据操作
     * @param numRows
     * @return
     */
    protected String getUpdateStatement(int numRows) {
        boolean first;
        StringBuilder sb = new StringBuilder();
        sb.append("INSERT INTO ");
        sb.append(tableName);
        sb.append("(");
        first = true;
        for (String column : columnNames) {
            if (first) {
                first = false;
            } else {
                sb.append(", ");
            }
            sb.append(column);
        }

        sb.append(") VALUES(");
        for (int i = 0; i             if (i > 0) {
                sb.append("),(");
            }
            for (int j = 0; j                 if (j > 0) {
                    sb.append(", ");
                }
                sb.append("?");
            }
        }

        sb.append(") ON DUPLICATE KEY UPDATE ");

        first = true;
        for (String column : columnNames) {
            if (first) {
                first = false;
            } else {
                sb.append(", ");
            }

            sb.append(column).append("=VALUES(").append(column).append(")");
        }

        String query = sb.toString();
        LOG.debug("Using upsert query: " + query);
        return query;
    }

    protected PreparedStatement getPreparedStatement(List userRecords) throws SQLException {

        PreparedStatement stmt = null;

        // Synchronize on connection to ensure this does not conflict
        // with the operations in the update thread.
        Connection cOnn= ConnectionRelate.getConn();
        stmt = conn.prepareStatement(getUpdateStatement(userRecords.size()));
        // Inject the record parameters into the UPDATE and WHERE clauses. This
        // assumes that the update key column is the last column serialized in
        // by the underlying record. Our code auto-gen process for exports was
        // responsible for taking care of this constraint.
        int i = 0;
        for (Writable record : userRecords) {
           setParam(stmt, i, record);
           i += columnNames.length;
        }
        stmt.addBatch();

        return stmt;
    }

 try {
            reader = new SequenceFile.Reader(FileSystem.get(new Configuration()), path, new Configuration());
            StringTriWritable key = new StringTriWritable();
            Writable value = (MediaScoreVector) reader.getValueClass().newInstance();
            while (reader.next(key, value)) {
               list.add(value);
               index++;
               if(index == batchSize){
                   index = 0;
                   //exp the data to the mysql
                   stmt = mediaSQL.getPreparedStatement(list);
                   stmt.executeBatch();

                   list.clear();
               }
            }

            //deal with size less batchsize
            if(list.size()>0) {
                stmt = mediaSQL.getPreparedStatement(list);
                stmt.executeBatch();

                list.clear();
                list = null;
            }


        } catch (Exception e) {
            。。。。
        } finally {
            if(stmt!=null){
                
               。。。。
            }
            。。。。
        }

至此说完。。。

本人才疏学浅，有错误之处请不吝指教。

关于insert的数量，很有必要深入研究下。