分析Hive数据

功能介绍

本小节介绍了如何使用样例程序完成分析任务。样例程序主要有以下方式。

• 使用JDBC接口提交数据分析任务。
• 使用HCatalog接口提交数据分析任务。

样例代码

• 使用Hive JDBC接口提交数据分析任务，参考样例程序中的JDBCExample.java。
  1. 定义HiveQL。HiveQL必须为单条语句，注意HiveQL不能包含“;”。

        // 定义HQL，不能包含“;” 
        String[] sqls = {"CREATE TABLE IF NOT EXISTS employees_info(id INT,name STRING)", 
                 "SELECT COUNT(*) FROM employees_info", "DROP TABLE employees_info"}; 

  2. 拼接JDBC URL。

     // 拼接JDBC URL
     StringBuilder sBuilder = new StringBuilder(
       "jdbc:hive2://").append(clientInfo.getZkQuorum()).append("/");
     
     if (isSecurityMode) {
         // 安全模式
         // ZooKeeper登录认证 
         sBuilder.append(";serviceDiscoveryMode=")
                 .append(clientInfo.getServiceDiscoveryMode())
                 .append(";zooKeeperNamespace=")
                 .append(clientInfo.getZooKeeperNamespace())
                 .append(";sasl.qop=")
                 .append(clientInfo.getSaslQop())
                 .append(";auth=")
                 .append(clientInfo.getAuth())
                 .append(";principal=")
                 .append(clientInfo.getPrincipal())
                 .append(";");
     } else {
         // 普通模式
         sBuilder.append(";serviceDiscoveryMode=")
                 .append(clientInfo.getServiceDiscoveryMode())
                 .append(";zooKeeperNamespace=")
                 .append(clientInfo.getZooKeeperNamespace())
                 .append(";auth=none");
     }
     String url = sBuilder.toString();

     以上是通过ZooKeeper的方式访问Hive。若直连HiveServer的方式访问Hive，需按如下方式拼接JDBC URL，并将hiveclient.properties文件中的zk.quorum配置项的端口改为10000。

     // 拼接JDBC URL
     StringBuilder sBuilder = new StringBuilder(
       "jdbc:hive2://").append(clientInfo.getZkQuorum()).append("/");
     
     if (isSecurityMode) {
         // 安全模式
         // ZooKeeper登录认证 
         sBuilder.append(";sasl.qop=")
                 .append(clientInfo.getSaslQop())
                 .append(";auth=")
                 .append(clientInfo.getAuth())
                 .append(";principal=")
                 .append(clientInfo.getPrincipal())
                 .append(";");
     } else {
         // 普通模式
         sBuilder.append(";auth=none");
     }
     String url = sBuilder.toString();

     注：直连HiveServer时，若当前连接的HiveServer故障则会导致访问Hive失败；若使用ZooKeeper的访问Hive，只要有任一个HiveServer实例可正常提供服务即可。因此使用JDBC时建议通过ZooKeeper的方式访问Hive。

  3. 加载Hive JDBC驱动。

        // 加载Hive JDBC驱动 
        Class.forName(HIVE_DRIVER);

  4. 填写正确的用户名，获取JDBC连接，确认HQL的类型（DDL/DML），调用对应的接口执行HiveQL，输出查询的列名和结果到控制台，关闭JDBC连接。

      
        Connection connection = null; 
          try { 
            // 获取JDBC连接 
            // 第二个参数需要填写正确的用户名，否则会以匿名用户(anonymous)登录
            connection = DriverManager.getConnection(url, "userName", ""); 
               
            // 建表 
            // 表建完之后，如果要往表中导数据，可以使用LOAD语句将数据导入表中，比如从HDFS上将数据导入表: 
            //load data inpath '/tmp/employees.txt' overwrite into table employees_info; 
            execDDL(connection,sqls[0]); 
            System.out.println("Create table success!"); 
              
            // 查询 
            execDML(connection,sqls[1]); 
               
            // 删表 
            execDDL(connection,sqls[2]); 
            System.out.println("Delete table success!"); 
          } 
          finally { 
            // 关闭JDBC连接 
            if (null != connection) { 
              connection.close(); 
            } 
       
      public static void execDDL(Connection connection, String sql) 
        throws SQLException { 
          PreparedStatement statement = null; 
          try { 
            statement = connection.prepareStatement(sql); 
            statement.execute(); 
          } 
          finally { 
            if (null != statement) { 
              statement.close(); 
            } 
          } 
        } 
       
       
        public static void execDML(Connection connection, String sql) throws SQLException { 
          PreparedStatement statement = null; 
          ResultSet resultSet = null; 
          ResultSetMetaData resultMetaData = null; 
           
          try { 
            // 执行HQL 
            statement = connection.prepareStatement(sql); 
            resultSet = statement.executeQuery(); 
             
            // 输出查询的列名到控制台 
            resultMetaData = resultSet.getMetaData(); 
            int columnCount = resultMetaData.getColumnCount(); 
            for (int i = 1; i <= columnCount; i++) { 
              System.out.print(resultMetaData.getColumnLabel(i) + '\t'); 
            } 
            System.out.println(); 
             
            // 输出查询结果到控制台 
            while (resultSet.next()) { 
              for (int i = 1; i <= columnCount; i++) { 
                System.out.print(resultSet.getString(i) + '\t'); 
              } 
              System.out.println(); 
            } 
          } 
          finally { 
            if (null != resultSet) { 
              resultSet.close(); 
            } 
             
            if (null != statement) { 
              statement.close(); 
            } 
          } 
        }     

• 使用HCatalog接口提交数据分析任务，参考样例程序中的HCatalogExample.java。
  1. 编写Map类，从Hive的表中获取数据。

        public static class Map extends
                 Mapper<LongWritable, HCatRecord, IntWritable, IntWritable> {
             int age;
             @Override
             protected void map(
                     LongWritable key,
                     HCatRecord value,
                     Context context)
                     throws IOException, InterruptedException {
                 age = (Integer) value.get(0);
                 context.write(new IntWritable(age), new IntWritable(1));
             }
         }     

  2. 编写Reduce类，对从Hive表中读取到的数据进行统计。

         public static class Reduce extends Reducer<IntWritable, IntWritable,
         IntWritable, HCatRecord> {
           @Override
           protected void reduce(
                   IntWritable key,
                   Iterable<IntWritable> values,
                   Context context)
                   throws IOException, InterruptedException {
               int sum = 0;
               Iterator<IntWritable> iter = values.iterator();
               while (iter.hasNext()) {
                   sum++;
                   iter.next();
               }
               HCatRecord record = new DefaultHCatRecord(2);
               record.set(0, key.get());
               record.set(1, sum);
     
               context.write(null, record);
             }
         } 

  3. 在run()方法中配置job后，执行main()方法，提交任务。

         public int run(String[] args) throws Exception {
           
             HiveConf.setLoadMetastoreConfig(true);
             Configuration conf = getConf();
             String[] otherArgs = args;
             
             String inputTableName = otherArgs[0];
             String outputTableName = otherArgs[1];
             String dbName = "default";
     
             @SuppressWarnings("deprecation")
             Job job = new Job(conf, "GroupByDemo");
             
             HCatInputFormat.setInput(job, dbName, inputTableName);
             job.setInputFormatClass(HCatInputFormat.class);
             job.setJarByClass(HCatalogExample.class);
             job.setMapperClass(Map.class);
             job.setReducerClass(Reduce.class);
             job.setMapOutputKeyClass(IntWritable.class);
             job.setMapOutputValueClass(IntWritable.class);
             job.setOutputKeyClass(WritableComparable.class);
             job.setOutputValueClass(DefaultHCatRecord.class);
             
             OutputJobInfo outputjobInfo = OutputJobInfo.create(dbName,outputTableName, null);
             HCatOutputFormat.setOutput(job, outputjobInfo);
             HCatSchema schema = outputjobInfo.getOutputSchema();
             HCatOutputFormat.setSchema(job, schema);
             job.setOutputFormatClass(HCatOutputFormat.class);
             
             return (job.waitForCompletion(true) ? 0 : 1);
         }
         public static void main(String[] args) throws Exception {
             int exitCode = ToolRunner.run(new HCatalogExample(), args);
             System.exit(exitCode);
         }