文档首页 > > 开发指南> MapReduce应用开发> 开发程序> MapReduce访问多组件样例程序> 样例代码说明

样例代码说明

分享
更新时间: 2019/04/30 GMT+08:00

功能介绍

该样例主要分为三个部分。

  • 从HDFS原文件中抽取name信息,查询HBase、Hive相关数据,并进行数据拼接,通过类MultiComponentMapper继承Mapper抽象类实现。
  • 获取拼接后的数据取最后一条输出到HBase、HDFS,通过类MultiComponentReducer继承Reducer抽象类实现。
  • main方法提供建立一个MapReduce job,并提交MapReduce作业到Hadoop集群。

代码样例

下面代码片段仅为演示,具体代码请参见com.huawei.bigdata.mapreduce.examples.MultiComponentExample类

样例1:类MultiComponentMapper定义Mapper抽象类的map方法。

private static class MultiComponentMapper extends Mapper<Object, Text, Text, Text> {

    Configuration conf;

    @Override protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {

      String name = "";
      String line = value.toString();

      //加载配置文件
        conf = context.getConfiguration();
        
        setJaasInfo("krb5.conf", "jaas.conf");
        LoginUtil.setJaasConf(ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME, "test", KEYTAB);
        LoginUtil.setZookeeperServerPrincipal(ZOOKEEPER_SERVER_PRINCIPAL_KEY, ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL);

        //准备hive query
        //加载parameter
        Properties clientInfo = null;
        InputStream fileInputStream = null;
        try {
            clientInfo = new Properties();
            File propertiesFile = new File(hiveClientProperties);
            fileInputStream = new FileInputStream(propertiesFile);
            clientInfo.load(fileInputStream);
        } catch (Exception e) {
        } finally {
            if (fileInputStream != null) {
                fileInputStream.close();
            }
        }

        String zkQuorum = clientInfo.getProperty("zk.quorum");
        String zooKeeperNamespace = clientInfo.getProperty("zooKeeperNamespace");
        String serviceDiscoveryMode = clientInfo.getProperty("serviceDiscoveryMode");

        // 创建Hive鉴权信息
        // Read this carefully:
        // MapReduce can only use Hive through JDBC.
        // Hive will submit another MapReduce Job to execute query.
        // So we run Hive in MapReduce is not recommended.
        final String driver = "org.apache.hive.jdbc.HiveDriver";

        String sql = "select name,sum(stayTime) as "
                + "stayTime from person where name = ? group by name";

        StringBuilder sBuilder = new StringBuilder("jdbc:hive2://").append(zkQuorum).append("/");
        // in map or reduce, use 'auth=delegationToken'
        sBuilder
                .append(";serviceDiscoveryMode=")
                .append(serviceDiscoveryMode)
                .append(";zooKeeperNamespace=")
                .append(zooKeeperNamespace)
                .append(";auth=delegationToken;");

        String url = sBuilder.toString();

        try {
            Class.forName(driver);
            hiveConn = DriverManager.getConnection(url, "", "");
            statement = hiveConn.prepareStatement(sql);
        } catch (Exception e) {
            LOG.error("Init jdbc driver failed.", e);
        }

        //创建hbase连接
        try {
            // Create a HBase connection
            hbaseConn = ConnectionFactory.createConnection(conf);
            // get table
            table = hbaseConn.getTable(TableName.valueOf(HBASE_TABLE_NAME));
        } catch (IOException e) {
            LOG.error("Exception occur when connect to HBase", e);
            throw e;
        }

      if (line.contains("male")) {
        name = line.substring(0, line.indexOf(","));
      }
      // 1. 读取HBase数据
      String hbaseData = readHBase();

      // 2. 读取Hive数据
      String hiveData = readHive(name);

      // Map输出键值对,内容为HBase与Hive数据拼接的字符串
      context.write(new Text(name), new Text("hbase:" + hbaseData + ", hive:" + hiveData));
    }

样例2:HBase数据读取的readHBase方法。

    private String readHBase() {
      String tableName = "table1";
      String columnFamily = "cf";
      String hbaseKey = "1";
      String hbaseValue;

      Configuration hbaseConfig = HBaseConfiguration.create(conf);
      org.apache.hadoop.hbase.client.Connection conn = null;
      try {

        // 创建一个HBase Get请求实例
        Get get = new Get(hbaseKey.getBytes());
        // 提交Get请求
        Result result = table.get(get);
        hbaseValue = Bytes.toString(result.getValue(columnFamily.getBytes(), "cid".getBytes()));

        return hbaseValue;

      } catch (IOException e) {
        LOG.warn("Exception occur ", e);
      } finally {
        if (hbaseConn != null) {
          try {
            hbaseConn.close();
          } catch (Exception e1) {
            LOG.error("Failed to close the connection ", e1);
          }
        }
      }

      return "";
    }

样例3:Hive数据读取的readHive方法。

    private int readHive(String name) {
      
      ResultSet resultSet = null;
      try {
        statement.setString(1, name);
        resultSet = statement.executeQuery();

        if (resultSet.next()) {
          return resultSet.getInt("stayTime"); 
        }       
      } catch (SQLException e) {
        LOG.warn("Exception occur ", e);
      } finally {
        if (null != resultSet) {
          try {
            resultSet.close();
          } catch (SQLException e) {
            // handle exception
          }
        }
        if (null != statement) {
          try {
            statement.close();
          } catch (SQLException e) {
            // handle exception
          }
        }
        if (null != hiveConn) {
          try {
            hiveConn.close();
          } catch (SQLException e) {
            // handle exception
          }
        }
      }

      return 0;
    }

样例4:类MultiComponentReducer定义Reducer抽象类的reduce方法。

    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
      
        Text finalValue = new Text("");

        setJaasInfo("krb5.conf", "jaas.conf");
        LoginUtil.setJaasConf(ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME, "test", KEYTAB);
        LoginUtil.setZookeeperServerPrincipal(ZOOKEEPER_SERVER_PRINCIPAL_KEY, ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL);

        conf = context.getConfiguration();
        try {
            // 创建hbase连接
            conn = ConnectionFactory.createConnection(conf);
            // 得到表
            table = conn.getTable(TableName.valueOf(HBASE_TABLE_NAME));
        } catch (IOException e) {
            LOG.error("Exception occur when connect to HBase", e);
            throw e;
        }
      
      for (Text value : values) {
        finalValue = value;
      }

      // 将结果输出到HBase
      writeHBase(key.toString(), finalValue.toString());

      // 将结果保存到HDFS
      context.write(key, finalValue);
    }

样例5:结果输出到HBase的writeHBase方法。

    private void writeHBase(String rowKey, String data) {
      
      try {
        // 创建一个HBase Put请求实例
        List<Put> list = new ArrayList<Put>();
        byte[] row = Bytes.toBytes("1");
        Put put = new Put(row);
        byte[] family = Bytes.toBytes("cf");
        byte[] qualifier = Bytes.toBytes("value");
        byte[] value = Bytes.toBytes(data);
        put.addColumn(family, qualifier, value);
        list.add(put);
        // 执行Put请求
        table.put(list);
      } catch (IOException e) {
        LOG.warn("Exception occur ", e);
      } finally {
        if (conn != null) {
          try {
            conn.close();
          } catch (Exception e1) {
            LOG.error("Failed to close the connection ", e1);
          }
        }
      }

    }

样例6:main()方法创建一个job,配置相关依赖,配置相关鉴权信息,提交作业到hadoop集群。

    public static void main(String[] args) throws Exception {
      
      // 清理所需目录
      MultiComponentExample.cleanupBeforeRun();

      // 查找Hive依赖jar包
      Class hiveDriverClass = Class.forName("org.apache.hive.jdbc.HiveDriver");
      Class thriftClass = Class.forName("org.apache.thrift.TException");
      Class thriftCLIClass = Class.forName("org.apache.hive.service.cli.thrift.TCLIService");
      Class hiveConfClass = Class.forName("org.apache.hadoop.hive.conf.HiveConf");
      Class hiveTransClass = Class.forName("org.apache.thrift.transport.HiveTSaslServerTransport");
      Class hiveMetaClass = Class.forName("org.apache.hadoop.hive.metastore.api.MetaException");
      Class hiveShimClass = Class.forName("org.apache.hadoop.hive.thrift.HadoopThriftAuthBridge23");

      // 添加Hive运行依赖到Job
      JarFinderUtil
          .addDependencyJars(config, hiveDriverClass, thriftCLIClass, thriftClass, hiveConfClass, hiveTransClass,
              hiveMetaClass, hiveShimClass);

      //开启Kerberos认证的安全集群登录
      if("kerberos".equalsIgnoreCase(config.get("hadoop.security.authentication"))){
      //security mode
          LoginUtil.setJaasConf(ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME, PRINCIPAL, KEYTAB);
          LoginUtil.setZookeeperServerPrincipal(ZOOKEEPER_SERVER_PRINCIPAL_KEY, ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL);
       System.setProperty("java.security.krb5.conf", KRB);
    LoginUtil.login(PRINCIPAL, KEYTAB, KRB, config);
    } 
      // 添加Hive配置文件
      config.addResource("hive-site.xml");
      // 添加HBase配置文件
      Configuration conf = HBaseConfiguration.create(config);

      // 实例化Job
      Job job = Job.getInstance(conf);
      job.setJarByClass(MultiComponentExample.class);

      // 设置mapper&reducer类
      job.setMapperClass(MultiComponentMapper.class);
      job.setReducerClass(MultiComponentReducer.class);

      //设置Job输入输出路径
      FileInputFormat.addInputPath(job, new Path(baseDir, INPUT_DIR_NAME + File.separator + "data.txt"));
      FileOutputFormat.setOutputPath(job, new Path(baseDir, OUTPUT_DIR_NAME));

      // 设置输出键值类型
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(Text.class);

      // HBase提供工具类添加HBase运行依赖到Job
      TableMapReduceUtil.addDependencyJars(job);

      // 安全模式下必须要执行这个操作
      // HBase添加鉴权信息到Job,map或reduce任务将会使用此处的鉴权信息
      TableMapReduceUtil.initCredentials(job);

      // 创建Hive鉴权信息
      Properties clientInfo = null;
      InputStream fileInputStream = null;
      try {
          clientInfo = new Properties();
          File propertiesFile = new File(hiveClientProperties);
          fileInputStream = new FileInputStream(propertiesFile);
          clientInfo.load(fileInputStream);
      } catch (Exception e) {
      } finally {
          if (fileInputStream != null) {
              fileInputStream.close();
          }
      }
      String zkQuorum = clientInfo.getProperty("zk.quorum");//zookeeper节点ip和端口列表
      String zooKeeperNamespace = clientInfo.getProperty("zooKeeperNamespace");
      String serviceDiscoveryMode = clientInfo.getProperty("serviceDiscoveryMode");
      String principal = clientInfo.getProperty("principal");
      String auth = clientInfo.getProperty("auth");
      String sasl_qop = clientInfo.getProperty("sasl.qop");
      StringBuilder sBuilder = new StringBuilder("jdbc:hive2://").append(zkQuorum).append("/");
      sBuilder.append(";serviceDiscoveryMode=").append(serviceDiscoveryMode).append(";zooKeeperNamespace=")
          .append(zooKeeperNamespace)
          .append(";sasl.qop=")
          .append(sasl_qop)
          .append(";auth=")
          .append(auth)
          .append(";principal=")
          .append(principal)
          .append(";");
      String url = sBuilder.toString();
      Connection connection = DriverManager.getConnection(url, "", "");
      String tokenStr = ((HiveConnection) connection)
          .getDelegationToken(UserGroupInformation.getCurrentUser().getShortUserName(), PRINCIPAL);
      connection.close();
      Token<DelegationTokenIdentifier> hive2Token = new Token<DelegationTokenIdentifier>();
      hive2Token.decodeFromUrlString(tokenStr);
      // 添加Hive鉴权信息到Job
      job.getCredentials().addToken(new Text("hive.server2.delegation.token"), hive2Token);
      job.getCredentials().addToken(new Text(HiveAuthFactory.HS2_CLIENT_TOKEN), hive2Token);

      // 提交作业
      System.exit(job.waitForCompletion(true) ? 0 : 1);

    }
说明:

样例中所有zkQuorum对象需替换为实际ZooKeeper集群节点信息。

分享:

    相关文档

    相关产品

文档是否有解决您的问题?

提交成功!

非常感谢您的反馈,我们会继续努力做到更好!

反馈提交失败,请稍后再试!

*必选

请至少选择或填写一项反馈信息

字符长度不能超过200

提交反馈 取消

如您有其它疑问,您也可以通过华为云社区问答频道来与我们联系探讨

跳转到云社区