代码实例

package com.epoint.com.mysql_mpp_full;

import java.io.FileNotFoundException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.TimeZone;

public class AutoMysqltoMPP {
    private static String MYSQLUSERNAME = "root";
    private static String MYSQLPASSWORD = "Gepoint";
    private static String MYSQLDRIVER = "com.mysql.jdbc.Driver";
    private static String MYSQLURL = "jdbc:mysql://100.2.5.221:3307/dep_fr_db";
    private static String MYSQLDATABASE = "dep_fr_db";

    private static String MPPDRIVER = "com.MPP.jdbc.Driver";
    private static String MPPURL = "jdbc:MPP://100.2.5.1:5258/dep_fr_db";
    private static String MPPUSERNAME = "mpp";
    private static String MPPPASSWORD = "h3c";

    private static Connection mysqlconn = null;
    private static Statement mysqlpstm = null;
    private static ResultSet mysqlrs = null;

    private static Connection mppconn = null;
    private static Statement mppstm = null;
    private static ResultSet mpprs = null;

    String sql1 = " ";
    String sql2 = " ";
    String sql3 = " ";
    String sql4 = " ";
    String sql5 = " ";
    String sql6 = " ";

    public static void main(String[] args) throws Exception {
        AutoMysqltoMPP aidth = new AutoMysqltoMPP();
        aidth.getMYSQLConnection();
        aidth.MYSQLReleaseResource();
        aidth.getMPPConnection();
        aidth.MPPReleaseResource();
        aidth.CreateMPPTable();
        // aidth.ImportDataToMPP();
        System.out.println("表已创建完毕，赶紧去查看表吧！！");

        SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss:SS");
        TimeZone t = sdf.getTimeZone();
        t.setRawOffset(0);
        sdf.setTimeZone(t);
        Long startTime = System.currentTimeMillis();
        // 此段为要放置测取时间的函数
        mysqlconn = getMYSQLConnection();
        mppconn = getMPPConnection();
        mppstm = mppconn.createStatement();
        mysqlpstm = mysqlconn.createStatement();
        mppstm.execute("TRUNCATE table code_main");
        aidth.tableInput();

        Long endTime = System.currentTimeMillis();
        System.out.println("用时：" + sdf.format(new Date(endTime - startTime)));
    }

    public void CreateMPPTable() {
        mysqlconn = getMYSQLConnection();
        mppconn = getMPPConnection();
        try {
            mppstm = mppconn.createStatement();
            mysqlpstm = mysqlconn.createStatement();
            int i = 0;
            String sql = "SELECT table_schema\r\n" + "  ,table_name\r\n" + "    ,(\r\n" + "     CASE \r\n"
                    + "         WHEN ORDINAL_POSITION = mincol\r\n" + "             AND ORDINAL_POSITION < maxcol\r\n"
                    + "             THEN CONCAT (\"create  table if not exists \"\r\n"
                    + "                     ,table_schema\r\n" + "                      ,\".\"\r\n"
                    + "                     ,table_name\r\n" + "                        ,\"(`\"\r\n"
                    + "                     ,column_name\r\n" + "                       ,\"` \"\r\n"
                    + "                     ,COLUMN_TYPE\r\n" + "                       ,\",\"\r\n"
                    + "                     )\r\n" + "          WHEN ORDINAL_POSITION = mincol\r\n"
                    + "             AND ORDINAL_POSITION = maxcol\r\n"
                    + "             THEN CONCAT (\"create  table if not exists \"\r\n"
                    + "                     ,table_schema\r\n" + "                      ,\".\"\r\n"
                    + "                     ,table_name\r\n" + "                        ,\"(`\"\r\n"
                    + "                     ,column_name\r\n" + "                       ,\"` \"\r\n"
                    + "                     ,COLUMN_TYPE\r\n" + "                       ,\");\"\r\n"
                    + "                     )\r\n" + "          WHEN ORDINAL_POSITION > mincol\r\n"
                    + "             AND ORDINAL_POSITION < maxcol\r\n" + "              THEN CONCAT (\r\n"
                    + "                     \"`\"\r\n" + "                      ,column_name\r\n"
                    + "                     ,\"` \"\r\n" + "                        ,COLUMN_TYPE\r\n"
                    + "                     ,\",\"\r\n" + "                     )\r\n"
                    + "         WHEN ORDINAL_POSITION = maxcol\r\n" + "             THEN CONCAT (\r\n"
                    + "                     \"`\"\r\n" + "                      ,column_name\r\n"
                    + "                     ,\"` \"\r\n" + "                        ,COLUMN_TYPE\r\n"
                    + "                     ,\");\"\r\n" + "                        )\r\n" + "          END\r\n"
                    + "     ) AS statement\r\n" + " ,ORDINAL_POSITION\r\n" + "  ,maxcol\r\n" + "    ,mincol\r\n"
                    + "FROM (\r\n" + "  SELECT b.table_schema,b.table_name,b.ORDINAL_POSITION,b.column_name,\r\n"
                    + " (case\r\n" + "  when column_type = 'timestamp' then 'datetime'\r\n"
                    + " when column_type = 'bit(1)' then 'int(1)'\r\n" + "  else\r\n" + "       column_type\r\n"
                    + " end ) AS column_type\r\n" + "       ,a.maxcol\r\n" + "      ,a.mincol\r\n" + "  FROM (\r\n"
                    + "     SELECT table_schema\r\n" + "            ,table_name\r\n"
                    + "         ,max(ORDINAL_POSITION) maxcol\r\n" + "          ,min(ORDINAL_POSITION) mincol\r\n"
                    + "     FROM information_schema.COLUMNS\r\n" + "        GROUP BY table_schema\r\n"
                    + "         ,table_name\r\n" + "        ) a\r\n" + "    JOIN (\r\n" + "     SELECT table_schema\r\n"
                    + "         ,table_name\r\n" + "            ,ORDINAL_POSITION\r\n" + "          ,column_name\r\n"
                    + "         ,COLUMN_TYPE\r\n" + "       FROM information_schema.COLUMNS\r\n"
                    + "     ORDER BY table_schema\r\n" + "          ,table_name\r\n"
                    + "         ,ORDINAL_POSITION ASC\r\n" + "      ) b ON a.table_schema = b.table_schema\r\n"
                    + "     AND a.table_name = b.table_name\r\n" + "    ) c\r\n" + "WHERE table_schema = '"
                    + MYSQLDATABASE + "'";
            mysqlrs = mysqlpstm.executeQuery(sql);
            while (mysqlrs.next()) {
                sql1 = mysqlrs.getString(3);
                sql2 = sql2 + sql1;
            }

            sql3 = "create database IF NOT EXISTS " + MYSQLDATABASE;
            mppstm.execute(sql3);
            System.out.println("-------------------建mpp表,表结构的语句为：" + sql2);
            String[] sqls = sql2.split(";");
            for (String m : sqls) {
                mppstm.execute(m);
            }
            System.out.println("----------------------------------------建mpp表已结束！！！！！！！！！！！！！！！！！！！！！！！！！！！！！");
            mppstm.close();
            mysqlpstm.close();
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            MYSQLReleaseResource();
            MPPReleaseResource();
        }
    }

    public void ImportDataToMPP() {
        mysqlconn = getMYSQLConnection();
        mppconn = getMPPConnection();
        String sql = "select table_name from user_tables where num_rows > 0 order by table_name asc";
        int i = 0;
        try {
            mysqlpstm = mysqlconn.createStatement();
            mysqlrs = mysqlpstm.executeQuery(sql);
            mppstm = mppconn.createStatement();
            while (mysqlrs.next()) {
                i = i + 1;
                String table_name = mysqlrs.getString("table_name").replaceAll("\\$", "");
                String sql7 = "insert into " + MYSQLDATABASE + "." + table_name + " select * from " + MYSQLDATABASE
                        + "_ex." + table_name;
                System.out.println("现在插入第" + i + "个表：" + sql7);
                mppstm.execute(sql7);
            }
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            MYSQLReleaseResource();
            MPPReleaseResource();
        }
    }

    public static List<List<String>> tableInput() throws FileNotFoundException, SQLException {
        List<List<String>> FindList = new ArrayList<List<String>>();
        mysqlconn = getMYSQLConnection();
        mppconn = getMPPConnection();
        mppstm = mppconn.createStatement();
        mysqlpstm = mysqlconn.createStatement();
        PreparedStatement pre = null;
        ResultSet resultSet = null;
        String sql = "SELECT CODEID,CODENAME,LEVNUM,CATEGORYNUM,description,isfromsoa  FROM code_main";
        try {
            pre = mysqlconn.prepareStatement(sql);
            resultSet = pre.executeQuery();
            String[] columu = { "CODEID","CODENAME","LEVNUM","CATEGORYNUM","description","isfromsoa"};
            int i = 0;
            while (resultSet.next()) {
                List<String> minList = new ArrayList<String>();
                for (String each : columu) {
                    minList.add(resultSet.getString(each));
                }
                FindList.add(minList);
                i++;
                if (i % 10000 == 0) { // 设置的每次提交大小为10000
                    executeManySql(FindList);
                    FindList.removeAll(FindList);
                    System.out.println(i);
                }
            }
            executeManySql(FindList);// 最后别忘了提交剩余的
            return FindList;
        } catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } finally {
            try {
                pre.close();// 关闭Statement
            } catch (SQLException e) {
                e.printStackTrace();
            }
            try {
                mppstm.close();
                mysqlpstm.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        return null;
    }

//executeManySql(FindList)函数如下，因为数据量比较大，所以我设置的每次提交大小为10000，这样就不会内存溢出了。

    public static void executeManySql(List<List<String>> FindList) throws SQLException {
        mysqlconn = getMYSQLConnection();
        mppconn = getMPPConnection();
        mppstm = mppconn.createStatement();
        mysqlpstm = mysqlconn.createStatement();
        mppconn.setAutoCommit(false);
        Statement stat = null;

        PreparedStatement pst = (PreparedStatement) mppconn
                .prepareStatement("insert into code_main values (?,?,?,?,?,?)");
        for (List<String> minList : FindList) {
            for (int i = 0; i < minList.size(); i++) {
                pst.setString(i + 1, minList.get(i));
            }
            // 把一个SQL命令加入命令列表
            pst.addBatch();
        }
        // 执行批量更新
        pst.executeBatch();
        // 语句执行完毕，提交本事务
        mppconn.commit();
        pst.close();
        mppstm.close();
        mysqlpstm.close();// 一定要记住关闭连接，不然mysql回应为too many connection自我保护而断开。
    }//同时我还设置了计时的函数，可以看到这个从数据抽取到完成数据迁移的时间。

    public static Connection getMYSQLConnection() {
        try {
            Class.forName(MYSQLDRIVER);
            mysqlconn = DriverManager.getConnection(MYSQLURL, MYSQLUSERNAME, MYSQLPASSWORD);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException("class not find !", e);
        } catch (SQLException e) {
            throw new RuntimeException("get connection error!", e);
        }

        return mysqlconn;
    }

    public void MYSQLReleaseResource() {
        if (mysqlrs != null) {
            try {
                mysqlrs.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        if (mysqlpstm != null) {
            try {
                mysqlpstm.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        if (mysqlconn != null) {
            try {
                mysqlconn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }

    public static Connection getMPPConnection() {
        try {
            Class.forName(MPPDRIVER);
            mppconn = DriverManager.getConnection(MPPURL, MPPUSERNAME, MPPPASSWORD);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException("class not find !", e);
        } catch (SQLException e) {
            throw new RuntimeException("get connection error!", e);
        }
        return mppconn;
    }

    public void MPPReleaseResource() {
        if (mpprs != null) {
            try {
                mpprs.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        if (mppstm != null) {
            try {
                mppstm.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        if (mppconn != null) {
            try {
                mppconn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }
}