`

spark读取hbase数据做分布式计算

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

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

代码如下,注释已经写详细:

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;
 
/**
 * <pre>
 *
 * spark hbase 测试
 *
 *  Created with IntelliJ IDEA.
 * User: zhangdonghao
 * Date: 14-1-26
 * Time: 上午9:24
 * To change this template use File | Settings | File Templates.
 * </pre>
 *
 * @author zhangdonghao
 */
public class HbaseTest implements Serializable {
 
    public Log log = LogFactory.getLog(HbaseTest.class);
 
    /**
     * 将scan编码,该方法copy自 org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil
     *
     * @param scan
     * @return
     * @throws IOException
     */
    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<ImmutableBytesWritable, Result> hBaseRDD = sc.newAPIHadoopRDD(conf,
                    TableInputFormat.class, ImmutableBytesWritable.class,
                    Result.class);
 
            //从result中取出用户的等级,并且每一个算一次
            JavaPairRDD<Integer, Integer> levels = hBaseRDD.map(
                    new PairFunction<Tuple2<ImmutableBytesWritable, Result>, Integer, Integer>() {
                        @Override
                        public Tuple2<Integer, Integer> call(
                                Tuple2<ImmutableBytesWritable, Result> immutableBytesWritableResultTuple2)
                                throws Exception {
                            byte[] o = immutableBytesWritableResultTuple2._2().getValue(
                                    Bytes.toBytes("info"), Bytes.toBytes("levelCode"));
                            if (o != null) {
                                return new Tuple2<Integer, Integer>(Bytes.toInt(o), 1);
                            }
                            return null;
                        }
                    });
 
            //数据累加
            JavaPairRDD<Integer, Integer> counts = levels.reduceByKey(new Function2<Integer, Integer, Integer>() {
                public Integer call(Integer i1, Integer i2) {
                    return i1 + i2;
                }
            });
             
            //打印出最终结果
            List<Tuple2<Integer, Integer>> output = counts.collect();
            for (Tuple2 tuple : output) {
                System.out.println(tuple._1 + ": " + tuple._2);
            }
 
        } catch (Exception e) {
            log.warn(e);
        }
 
    }
 
    /**
     * spark如果计算没写在main里面,实现的类必须继承Serializable接口,<br>
     * </>否则会报 Task not serializable: java.io.NotSerializableException 异常
     */
    public static void main(String[] args) throws InterruptedException {
 
        new HbaseTest().start();
 
        System.exit(0);
    }
}


运行结果如下:

0: 28528
11: 708
4: 28656
2: 36315
6: 23848
8: 19802
10: 6913
9: 15988
3: 31950
1: 38872
7: 21600
5: 27190
12: 17
分享到:
评论

相关推荐

    spark使用java读取hbase数据做分布式计算.pdf

    总的来说,这个Java程序展示了如何使用Spark读取HBase数据并进行分布式计算。通过Spark的并行处理能力,可以高效地处理大规模的HBase数据,进行复杂的分析任务。要注意的是,实际应用中还需要考虑错误处理、资源管理...

    hbase-rdd:Spark RDD从HBase读取,写入和删除

    Spark 提供了高性能的分布式计算能力,而 HBase 则是针对大规模数据存储的列式数据库。本篇文章将详细探讨如何使用 Scala 和 Spark 的 Resilient Distributed Datasets (RDDs) 与 HBase 进行交互,包括读取、写入...

    Spark以及hbase学习资料

    在实际项目中,Spark可以用来做数据预处理、特征工程和模型训练,而HBase则作为数据仓库,存储大量的历史数据。这样的组合使得数据分析更加灵活,且能够应对海量数据的挑战。学习Spark和HBase,你需要掌握如何配置和...

    Kafka集成Spark Streaming并写入数据到HBase

    Spark Streaming则提供了基于微批处理的实时计算框架,能够高效地处理持续的数据流;而HBase作为分布式列式数据库,适合存储海量结构化半结构化数据。这三者之间的集成,使得实时数据流能够被快速处理并持久化存储。...

    Hadoop进行分布式计算的入门资料

    - Hadoop生态:包括Hive(数据仓库工具)、Pig(数据分析工具)、Spark(快速数据处理框架)等,它们与Hadoop协同工作,提供了更丰富的数据处理功能。 - HBase:基于Hadoop的分布式NoSQL数据库,适用于实时查询...

    spark-2.4.0-hive-hbase-Api.7z

    3. 从HBase中读取数据,转换成Spark DataFrame,方便进一步的数据处理和分析。 4. 利用Spark的MLlib库进行机器学习模型训练,预测和分类等任务。 在实际项目中,这些组件的集成可以帮助企业构建大规模数据处理平台...

    SparkHadoopHbase案例

    Hadoop是Apache基金会开发的一个开源项目,核心包括HDFS(分布式文件系统)和MapReduce(分布式计算模型)。HDFS提供了高容错性的数据存储,MapReduce则用于处理和生成大数据集。Hadoop是Spark的基础,Spark可以运行...

    hbase海量数据的全量导入方法

    3. **并行导入**:利用MapReduce或Spark等分布式计算框架,将数据分割成多个部分并行导入,进一步加速导入过程。 4. **优化HBase配置**:调整HBase的参数,如关闭自动compaction、增加memstore大小、调整flush和...

    Spark读取HbaseRDD的过程及源码解析

    通过以上步骤,Spark 能够根据用户配置高效地读取 HBase 数据并将其转换为 RDD,从而支持进一步的分布式计算。这种读取方式结合了 Spark 的并行处理能力和 HBase 的分布式存储特性,为大数据分析提供了强大支持。在...

    基于Spark的分布式车流量检测方法设计与实现.pdf

    1. 分布式计算框架:文件中提到的分布式计算框架指的是Spark。Spark是一个开源的分布式计算系统,它提供了一个快速且通用的计算引擎。Spark的核心是弹性分布式数据集(RDD),一个容错的并行数据结构,可以将数据...

    建立Hive和Hbase的映射关系,通过Spark将Hive表中数据导入ClickHouse

    接下来,Spark作为一个分布式计算框架,提供了一种灵活且高性能的方式来处理数据。在Hive和ClickHouse之间传输数据时,Spark可以作为一个中间层,利用其强大的数据处理能力,将Hive表的数据转换为适合ClickHouse存储...

    基于hadoop+hbase+springboot实现分布式网盘系统.zip

    总的来说,基于Hadoop、HBase和SpringBoot的分布式网盘系统利用了分布式存储和计算的优势,能够处理海量文件,支持高并发的读写操作,同时提供灵活的数据管理和查询功能。这个系统的设计和实现涉及到分布式系统设计...

    Spark大数据处理技术.pdf

    1. 数据准备:使用Spark的数据读取接口,将存储在HDFS、HBase、S3等存储系统中的数据读入Spark。 2. 数据转换:对数据进行各种转换操作,比如映射(map)、过滤(filter)、聚合(reduce)、连接(join)等。 3. 数据分析:...

    基于hbase + spark 实现常用推荐算法(主要用于精准广告投放和推荐系统).zip

    通过这个项目,学生可以深入了解大数据处理、分布式计算以及推荐系统的设计与实现,同时掌握HBase和Spark的实战应用。对于计算机专业毕业设计、课程设计以及项目开发来说,这是一个很好的实践课题,能够锻炼实际操作...

    基于spark+flume+kafka+hbase的实时日志处理分析系统.zip

    这个项目利用了Apache软件基金会的几个关键组件,包括分布式计算框架Spark、数据流传输工具Flume、消息中间件Kafka以及分布式数据库HBase。 1. **Apache Spark**:Spark是用于大规模数据处理的快速、通用和可扩展的...

    分布式计算开源框架Hadoop介绍.docx

    Hadoop 是一个由 Apache 开源基金会开发的分布式计算框架,被广泛应用于处理和存储海量数据。该框架的核心设计理念是 MapReduce 和 HDFS(Hadoop 分布式文件系统)。Hadoop 在众多大型互联网公司如亚马逊、Facebook ...

    spark解析csv文件,存入数据库

    对于处理CSV文件这样的文本数据,Spark提供了非常高效和易于使用的工具集,而不需要深入了解底层的数据存储和分布式计算的细节。通过Spark,即使是复杂的转换和数据清洗过程也能被简化为简单的API调用,大大提高了...

    【SparkCore篇07】RDD数据读取和保存1

    在Spark中,RDD(Resilient Distributed Datasets)是其核心的数据抽象,它是一种容错的、不可变的数据集合,可以在集群中分布式计算。本文将详细讨论RDD数据的读取和保存,特别是涉及JSON和SequenceFile这两种文件...

    基于spark streaming和kafka,hbase的日志统计分析系统.zip

    Spark Streaming通过微批处理来模拟流处理,将数据流划分为小批量的DStreams(Discretized Streams),并利用Spark Core的强大计算能力进行处理。这种设计使得Spark Streaming能够快速响应数据变化,同时保持与...

Global site tag (gtag.js) - Google Analytics