HBase——使用Put迁移MySql数据到Hbase-阿里云开发者社区

开发者社区> soledad_lhc> 正文

HBase——使用Put迁移MySql数据到Hbase

简介: 先上code: /** * 功能:迁移mysql上电池历史数据到hbase * Created by liuhuichao on 2016/12/6. */ public class MySqlToHBase { /** * 获取表 * @param...
+关注继续查看



先上code:



/**
 * 功能:迁移mysql上电池历史数据到hbase
 * Created by liuhuichao on 2016/12/6.
 */
public class MySqlToHBase {

    /**
     * 获取表
     * @param tableName
     * @return
     * @throws IOException
     */
    private  HTable connectHBase(String tableName) throws IOException{
        HTable table=null;
        Configuration conf= HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","lhc-centos");
        table=new HTable(conf,tableName);
        return table;
    }

    /**
     * 获取mysql连接
     * @return
     * @throws Exception
     */
    private  java.sql.Connection connectDB() throws  Exception{
        String userName="root";
        String password="root";
        String url="jdbc:mysql://10.0.1.42:3306/energy?useUnicode=true&characterEncoding=UTF-8";
        Class.forName("com.mysql.jdbc.Driver").newInstance();
        java.sql.Connection conn=DriverManager .getConnection(url,userName,password);
        return  conn;
    }

    /**
     * 将电池历史数据表中输入导入HBase
     * @throws Exception
     */
    @Test
    public void exportFromMySqlToHBase() throws Exception{
        java.sql.Connection dbConn=null;
        HTable table=null;

        Statement stmt=null;
        String strQuery="SELECT * FROM `res_battery_data_history` limit 10000,1000000";
        dbConn=connectDB();//连接mysql
        table=connectHBase("batteryDataHistory");//连接HBase
        try{

            stmt=dbConn.createStatement();
            ResultSet rs=stmt.executeQuery(strQuery);
            long beginTime=System.currentTimeMillis();//开始
            System.out.println( "beginTime:---"+beginTime);
            while (rs.next()) {
                UUID uuid = UUID.randomUUID();
                String rowKey = uuid.toString();//作为行健
                Put put = new Put(Bytes.toBytes(rowKey));
                /**
                 * family:baseData
                 */
                // Integer id=rs.getInt("id");
                String batteryNo = rs.getString("battery_no");
                Integer batteryType = rs.getInt("battery_type");
                Float voltageDeviation = rs.getFloat("voltage_deviation");
                Float totalVoltage = rs.getFloat("total_voltage");
                Float temprature1 = rs.getFloat("temprature1");
                Float temprature2 = rs.getFloat("temprature2");
                Float chargeNum = rs.getFloat("charge_num");
                Float longtitude = rs.getFloat("longtitude");
                Float latitude = rs.getFloat("latitude");
                Float totalCurrent = rs.getFloat("total_current");
                Float soc = rs.getFloat("soc");

                put.add(Bytes.toBytes("baseData"), Bytes.toBytes("batteryNo"), Bytes.toBytes(batteryNo));
                put.add(Bytes.toBytes("baseData"), Bytes.toBytes("batteryType"), Bytes.toBytes(batteryType));
                put.add(Bytes.toBytes("baseData"), Bytes.toBytes("voltageDeviation"), Bytes.toBytes(voltageDeviation));
                put.add(Bytes.toBytes("baseData"), Bytes.toBytes("totalVoltage"), Bytes.toBytes(totalVoltage));
                put.add(Bytes.toBytes("baseData"), Bytes.toBytes("temprature1"), Bytes.toBytes(temprature1));
                put.add(Bytes.toBytes("baseData"), Bytes.toBytes("temprature2"), Bytes.toBytes(temprature2));
                put.add(Bytes.toBytes("baseData"), Bytes.toBytes("chargeNum"), Bytes.toBytes(chargeNum));
                put.add(Bytes.toBytes("baseData"), Bytes.toBytes("longtitude"), Bytes.toBytes(longtitude));
                put.add(Bytes.toBytes("baseData"), Bytes.toBytes("latitude"), Bytes.toBytes(latitude));
                put.add(Bytes.toBytes("baseData"), Bytes.toBytes("totalCurrent"), Bytes.toBytes(totalCurrent));
                put.add(Bytes.toBytes("baseData"), Bytes.toBytes("soc"), Bytes.toBytes(soc));
                /**
                 * family:volumnData
                 */
                Float vol1 = rs.getFloat("vol1");
                Float vol2 = rs.getFloat("vol2");
                Float vol3 = rs.getFloat("vol3");
                Float vol4 = rs.getFloat("vol4");
                Float vol5 = rs.getFloat("vol5");
                Float vol6 = rs.getFloat("vol6");
                Float vol7 = rs.getFloat("vol7");
                Float vol8 = rs.getFloat("vol8");
                Float vol9 = rs.getFloat("vol9");
                Float vol10 = rs.getFloat("vol10");
                Float vol11 = rs.getFloat("vol11");
                Float vol12 = rs.getFloat("vol12");
                Float vol13 = rs.getFloat("vol13");
                Float vol14 = rs.getFloat("vol14");
                Float vol15 = rs.getFloat("vol15");
                Float vol16 = rs.getFloat("vol16");
                Float vol17 = rs.getFloat("vol17");
                Float vol18 = rs.getFloat("vol18");
                Float vol19 = rs.getFloat("vol19");
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v1"), Bytes.toBytes(vol1));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v2"), Bytes.toBytes(vol2));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v3"), Bytes.toBytes(vol3));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v4"), Bytes.toBytes(vol4));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v5"), Bytes.toBytes(vol5));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v6"), Bytes.toBytes(vol6));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v7"), Bytes.toBytes(vol7));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v8"), Bytes.toBytes(vol8));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v9"), Bytes.toBytes(vol9));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v10"), Bytes.toBytes(vol10));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v11"), Bytes.toBytes(vol11));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v12"), Bytes.toBytes(vol12));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v13"), Bytes.toBytes(vol13));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v14"), Bytes.toBytes(vol14));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v15"), Bytes.toBytes(vol15));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v16"), Bytes.toBytes(vol16));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v17"), Bytes.toBytes(vol17));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v18"), Bytes.toBytes(vol18));
                put.add(Bytes.toBytes("volumnData"), Bytes.toBytes("v19"), Bytes.toBytes(vol19));
                /**
                 * family:extraData
                 */
                String remarks = rs.getString("remarks");
                String testUserName = rs.getString("test_user_name");
                String createTime = rs.getString("create_time");
                Integer createUser = rs.getInt("create_user");
                Integer source = rs.getInt("source");
                put.add(Bytes.toBytes("extraData"), Bytes.toBytes("remarks"), Bytes.toBytes(remarks));
                put.add(Bytes.toBytes("extraData"), Bytes.toBytes("testUserName"), Bytes.toBytes(testUserName));
                put.add(Bytes.toBytes("extraData"), Bytes.toBytes("createTime"), Bytes.toBytes(createTime));
                put.add(Bytes.toBytes("extraData"), Bytes.toBytes("createUser"), Bytes.toBytes(createUser));
                put.add(Bytes.toBytes("extraData"), Bytes.toBytes("source"), Bytes.toBytes(source));
                table.put(put);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        finally {
            try{
                if(stmt !=null){
                    stmt.close();
                }
                if(dbConn !=null){
                    dbConn.close();
                }
                if(table!=null){
                    table.close();
                }
                long endTime=System.currentTimeMillis();//开始
                System.out.println( "endTime:---"+endTime);
            }catch(Exception e){
                e.printStackTrace();
            }
        }








    }
}




   结果:10000 row(s) in 218.0310 seconds

      (我是单机版的HBase,未使用HDFS做底层存储。)

     刚开始的sql没有加limit,直接上来把测试表里面的三百多万条直接插出来,结果还没查完,就堆栈溢出了;额。。。后来先塞了一万条到HBase,额,还好,速度很快,但是感觉这种导入方式不具有通用性,对于大量数据的导入是不ok的。






版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
[JDBC] Kettle on MaxCompute 使用指南
Kettle是一款开源的ETL工具,纯Java实现,可以在Windows、Unix和Linux上运行,提供图形化的操作界面,可以通过拖拽控件的方式,方便地定义数据传输的拓扑 。基本讲介绍基于Kettle的MaxCompute插件实现数据上云。
1551 0
阿里云城市服务商分享怎么使用DTS迁移PPAS数据
使用数据传输服务(DTS)将本地数据库迁移到 RDS for PPAS,可以实现应用不停服务的情况下,平滑完成数据库的迁移工作。迁移过程中,对本地的 Oracle 数据库没有影响。那么如何操作呢,湖北区域阿里云城市服务商捷讯技术告诉您怎么使用怎么使用DTS迁移PPAS数据
1467 0
使用EMR-Kafka Connect进行数据迁移
流式处理中经常会遇到Kafka与其他系统进行数据同步或者Kafka集群间数据迁移的情景。使用EMR Kafka Connect可以方便快速的实现数据同步或者数据迁移。本文介绍使用EMR Kafka Connect的REST API接口在Kafka集群间进行数据迁移。
13302 0
阿里云MySQL云数据库和Redis6.0数据库使用体验
阿里云MySQL云数据库和Redis6.0数据库使用体验
1616 0
蚂蚁集团自研数据库 OceanBase 在中华财险业务系统投入使用
中华财险车险系统已经在全国 29 省成功上线蚂蚁集团自研数据库 OceanBase。
362 0
阿里云RDS mysql配置及迁移数据方法新手教程
阿里云RDS mysql配置及迁移数据方法新手教程
1077 0
+关注
soledad_lhc
csdn博客:http://blog.csdn.net/lhc1105
508
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载