由于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
分享到:
相关推荐
总的来说,这个Java程序展示了如何使用Spark读取HBase数据并进行分布式计算。通过Spark的并行处理能力,可以高效地处理大规模的HBase数据,进行复杂的分析任务。要注意的是,实际应用中还需要考虑错误处理、资源管理...
Spark 提供了高性能的分布式计算能力,而 HBase 则是针对大规模数据存储的列式数据库。本篇文章将详细探讨如何使用 Scala 和 Spark 的 Resilient Distributed Datasets (RDDs) 与 HBase 进行交互,包括读取、写入...
在实际项目中,Spark可以用来做数据预处理、特征工程和模型训练,而HBase则作为数据仓库,存储大量的历史数据。这样的组合使得数据分析更加灵活,且能够应对海量数据的挑战。学习Spark和HBase,你需要掌握如何配置和...
Spark Streaming则提供了基于微批处理的实时计算框架,能够高效地处理持续的数据流;而HBase作为分布式列式数据库,适合存储海量结构化半结构化数据。这三者之间的集成,使得实时数据流能够被快速处理并持久化存储。...
- Hadoop生态:包括Hive(数据仓库工具)、Pig(数据分析工具)、Spark(快速数据处理框架)等,它们与Hadoop协同工作,提供了更丰富的数据处理功能。 - HBase:基于Hadoop的分布式NoSQL数据库,适用于实时查询...
3. 从HBase中读取数据,转换成Spark DataFrame,方便进一步的数据处理和分析。 4. 利用Spark的MLlib库进行机器学习模型训练,预测和分类等任务。 在实际项目中,这些组件的集成可以帮助企业构建大规模数据处理平台...
Hadoop是Apache基金会开发的一个开源项目,核心包括HDFS(分布式文件系统)和MapReduce(分布式计算模型)。HDFS提供了高容错性的数据存储,MapReduce则用于处理和生成大数据集。Hadoop是Spark的基础,Spark可以运行...
3. **并行导入**:利用MapReduce或Spark等分布式计算框架,将数据分割成多个部分并行导入,进一步加速导入过程。 4. **优化HBase配置**:调整HBase的参数,如关闭自动compaction、增加memstore大小、调整flush和...
通过以上步骤,Spark 能够根据用户配置高效地读取 HBase 数据并将其转换为 RDD,从而支持进一步的分布式计算。这种读取方式结合了 Spark 的并行处理能力和 HBase 的分布式存储特性,为大数据分析提供了强大支持。在...
1. 分布式计算框架:文件中提到的分布式计算框架指的是Spark。Spark是一个开源的分布式计算系统,它提供了一个快速且通用的计算引擎。Spark的核心是弹性分布式数据集(RDD),一个容错的并行数据结构,可以将数据...
接下来,Spark作为一个分布式计算框架,提供了一种灵活且高性能的方式来处理数据。在Hive和ClickHouse之间传输数据时,Spark可以作为一个中间层,利用其强大的数据处理能力,将Hive表的数据转换为适合ClickHouse存储...
总的来说,基于Hadoop、HBase和SpringBoot的分布式网盘系统利用了分布式存储和计算的优势,能够处理海量文件,支持高并发的读写操作,同时提供灵活的数据管理和查询功能。这个系统的设计和实现涉及到分布式系统设计...
1. 数据准备:使用Spark的数据读取接口,将存储在HDFS、HBase、S3等存储系统中的数据读入Spark。 2. 数据转换:对数据进行各种转换操作,比如映射(map)、过滤(filter)、聚合(reduce)、连接(join)等。 3. 数据分析:...
通过这个项目,学生可以深入了解大数据处理、分布式计算以及推荐系统的设计与实现,同时掌握HBase和Spark的实战应用。对于计算机专业毕业设计、课程设计以及项目开发来说,这是一个很好的实践课题,能够锻炼实际操作...
这个项目利用了Apache软件基金会的几个关键组件,包括分布式计算框架Spark、数据流传输工具Flume、消息中间件Kafka以及分布式数据库HBase。 1. **Apache Spark**:Spark是用于大规模数据处理的快速、通用和可扩展的...
Hadoop 是一个由 Apache 开源基金会开发的分布式计算框架,被广泛应用于处理和存储海量数据。该框架的核心设计理念是 MapReduce 和 HDFS(Hadoop 分布式文件系统)。Hadoop 在众多大型互联网公司如亚马逊、Facebook ...
对于处理CSV文件这样的文本数据,Spark提供了非常高效和易于使用的工具集,而不需要深入了解底层的数据存储和分布式计算的细节。通过Spark,即使是复杂的转换和数据清洗过程也能被简化为简单的API调用,大大提高了...
在Spark中,RDD(Resilient Distributed Datasets)是其核心的数据抽象,它是一种容错的、不可变的数据集合,可以在集群中分布式计算。本文将详细讨论RDD数据的读取和保存,特别是涉及JSON和SequenceFile这两种文件...
Spark Streaming通过微批处理来模拟流处理,将数据流划分为小批量的DStreams(Discretized Streams),并利用Spark Core的强大计算能力进行处理。这种设计使得Spark Streaming能够快速响应数据变化,同时保持与...