设为首页 加入收藏

TOP

spark使用java读取hbase数据做分布式计算
2018-12-07 01:59:59 】 浏览:8
Tags:spark 使用 java 读取 hbase 数据 分布式 计算

问题导读:
1.如何初始化sparkContext?
2.如何设置查询条件?
3.如何获得hbase查询结果Result?


由于spark提供的hbaseTest是scala版本,并没有提供java版。我将scala版本改为java版本,并根据数据做了些计算操作。

程序目的:查询出hbase满足条件的用户,统计各个等级个数。

代码如下,西面使用的hbase是0.94注释已经写详细:


package com.sdyc.ndspark.sys;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.util.Base64;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.Serializable;
import java.util.List;


public class HbaseTest implements Serializable {

public Log log = LogFactory.getLog(HbaseTest.class);


static String convertScanToString(Scan scan) throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(out);
scan.write(dos);
return Base64.encodeBytes(out.toByteArray());
}

public void start() {
//初始化sparkContext,这里必须在jars参数里面放上Hbase的jar,
// 否则会报unread block data异常
JavaSparkContext sc = new JavaSparkContext("spark://nowledgedata-n3:7077", "hbaseTest",
"/home/hadoop/software/spark-0.8.1",
new String[]{"target/ndspark.jar", "target\\dependency\\hbase-0.94.6.jar"});

//使用HBaseConfiguration.create()生成Configuration
// 必须在项目classpath下放上hadoop以及hbase的配置文件。
Configuration conf = HBaseConfiguration.create();
//设置查询条件,这里值返回用户的等级
Scan scan = new Scan();
scan.setStartRow(Bytes.toBytes("195861-1035177490"));
scan.setStopRow(Bytes.toBytes("195861-1072173147"));
scan.addFamily(Bytes.toBytes("info"));
scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("levelCode"));

try {
//需要读取的hbase表名
String tableName = "usertable";
conf.set(TableInputFormat.INPUT_TABLE, tableName);
conf.set(TableInputFormat.SCAN, convertScanToString(scan));

//获得hbase查询结果Result
JavaPairRDD hBaseRDD = sc.newAPIHadoopRDD(conf,
TableInputFormat.class, ImmutableBytesWritable.class,
Result.class);

//从result中取出用户的等级,并且每一个算一次
JavaPairRDD levels = hBaseRDD.map(
new PairFunction, Integer, Integer>() {
@Override
public Tuple2 call(
Tuple2 immutableBytesWritableResultTuple2)
throws Exception {
byte[] o = immutableBytesWritableResultTuple2._2().getValue(
Bytes.toBytes("info"), Bytes.toBytes("levelCode"));
if (o != null) {


编程开发网
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇Hbase Mapreduce示例:全库扫描(.. 下一篇HBase命令行无法删除命令的问题

评论

帐  号: 密码: (新用户注册)
验 证 码:
表  情:
内  容:

array(4) { ["type"]=> int(8) ["message"]=> string(24) "Undefined variable: jobs" ["file"]=> string(32) "/mnt/wp/cppentry/do/bencandy.php" ["line"]=> int(214) }