MySQL数据导入Hive-Java

每天学Java

发布于 2020-06-02 10:21:42

2.2K0

发布于 2020-06-02 10:21:42

文章被收录于专栏：每天学Java

文章来源:http://www.study-java.cn/
上一篇文章我们使用通过beeline执行一些常规的HQL，今天这一篇文章主要来看如果通过Java将MySQL数据导入到Hive中。

Sqoop

Sqoop并不在这篇文章的范围内，拿出来说的原因是，公司数据研发部门是通过Sqoop将数据库数据导入到Hive中，其原理是将数据库数据导入到HDFS中临时存储，然后在将文件导入到Hive中，最终删掉临时存储的文件。

而笔者并没有采用这种方式，原因很简单，我的目的是学习Hive，过多的用这些工具会增加了我的学习工具成本，所以我看了Sqoop的原理后，准备模仿一下，简单的实现数据的导入，过程如下:

连接MySQL
查询导入的数据
调用Hadoop的API将数据存入到HDFS中
将HDFS文件导入到Hive中

查询MySQL数据

这里我查询用户表的用户名称，年，月，日，并将结果集存入ResultSet中

        String driverclass = "com.mysql.jdbc.Driver";
        String url = "jdbc:mysql://IP:3306/xcxvt?useSSL=false";
        String username = "root";
        String passwd = "pwd";
        //连接
        try {
            Class.forName(driverclass);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
            System.exit(1);
        }
        Connection con = DriverManager.getConnection(url, username, passwd);
        Statement stmt = con.createStatement();
        //查询
        StringBuffer sql = new StringBuffer();
        sql.append("select userName, ");
        sql.append("substring(create_time,1,4) as year,");
        sql.append("substring(create_time,6,2) as month,");
        sql.append("substring(create_time,9,2) as day ");
        sql.append("from user_info");
        ResultSet resultSet = stmt.executeQuery(sql.toString());
        StringBuffer data = new StringBuffer();

存入HDFS

        //创建临时文件
        BufferedWriter bw = null;
        File file = null;
        try {
            file = File.createTempFile("userInfo", ".txt");
            bw = new BufferedWriter(new FileWriter(file));
        } catch (IOException e) {
            e.printStackTrace();
        }
        //写入字段，逗号隔开，这里和建表的字段分隔符一致
        bw.write("user_name,year,month,month");
        while (resultSet.next()) {
            data.setLength(0);
            data.append(resultSet.getString(1) + ",");
            data.append(resultSet.getString(2) + ",");
            data.append(resultSet.getString(3) + ",");
            data.append(resultSet.getString(4) + "\n");
            try {
                bw.write(data.toString());
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        try {
            bw.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://localhost:9000/"), new Configuration(), "chenlong");
        fileSystem.copyFromLocalFile(false, true, new Path(file.getPath()), new Path("/mysql"));
        //删除临时文件
        file.deleteOnExit();

导入Hive

        String driverName = "org.apache.hive.jdbc.HiveDriver";
        try {
            Class.forName(driverName);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
            System.exit(1);
        }
        Connection con = DriverManager.getConnection("jdbc:hive2://127.0.0.1:12000/xcx_user_db", "chenlong", "");
        Statement stmt = con.createStatement();
        //导入非分区数据
        String importData = "load data inpath '/mysql' overwrite into table user_unpartition";
        int count = stmt.executeUpdate(importData);

到这里实现数据的导入