大家可能都知道很熟悉Spark的两种常见的数据读取方式（存放到RDD中）：（1）、调用parallelize函数直接从集合中获取数据，并存入RDD中；Java版本如下：

JavaRDD<Integer> myRDD = sc.parallelize(Arrays.asList(1,2,3));

Scala版本如下：

val myRDD= sc.parallelize(List(1,2,3))

　　这种方式很简单，很容易就可以将一个集合中的数据变成RDD的初始化值；更常见的是（2）、从文本中读取数据到RDD中，这个文本可以是纯文本文件、可以是sequence文件；可以存放在本地(file://)、可以存放在HDFS（hdfs://）上，还可以存放在S3上。其实对文件来说，Spark支持Hadoop所支持的所有文件类型和文件存放位置。Java版如下：

/////////////////////////////////////////////////////////////////////
 User: 过往记忆
 Date: 14-6-29
 Time: 23:59
 bolg:
 本文地址：/archives/1051
 过往记忆博客，专注于hadoop、hive、spark、shark、flume的技术博客，大量的干货
 过往记忆博客微信公共帐号：iteblog_hadoop
/////////////////////////////////////////////////////////////////////
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
 
SparkConf conf = new SparkConf().setAppName("Simple Application");
JavaSparkContext sc = new JavaSparkContext(conf);
sc.addFile("wyp.data");
JavaRDD<String> lines = sc.textFile(SparkFiles.get("wyp.data"));

Scala版本如下：

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
 
val conf = new SparkConf().setAppName("Simple Application")
val sc = new SparkContext(conf)
sc.addFile("spam.data")
val inFile = sc.textFile(SparkFiles.get("spam.data"))

　　在实际情况下，我们需要的数据可能不是简单的存放在HDFS文本中，我们需要的数据可能就存放在Hbase中，那么我们如何用Spark来读取Hbase中的数据呢？本文的所有测试是基于Hadoop 2.2.0、Hbase 0.98.2、Spark 0.9.1，不同版本可能代码的编写有点不同。本文只是简单地用Spark来读取Hbase中的数据，如果需要对Hbase进行更强的操作，本文可能不能帮你。话不多说，Spark操作Hbase的Java版本代码如下：

package com.iteblog.spark;

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.protobuf.ProtobufUtil;

import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;

import org.apache.hadoop.hbase.util.Base64;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.spark.SparkConf;

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.Serializable;

import scala.Tuple2;

import java.io.IOException;

import java.util.List;

/**

 * User: iteblog

 * Date: 14-6-27

 * Time: 下午5:18

 *blog: http://www.iteblog.com

 *

 * Usage: bin/spark-submit --master yarn-cluster --class com.iteblog.spark.SparkFromHbase

 * --jars /home/q/hbase/hbase-0.96.0-hadoop2/lib/htrace-core-2.01.jar,

 * /home/q/hbase/hbase-0.96.0-hadoop2/lib/hbase-common-0.96.0-hadoop2.jar,

 * /home/q/hbase/hbase-0.96.0-hadoop2/lib/hbase-client-0.96.0-hadoop2.jar,

 * /home/q/hbase/hbase-0.96.0-hadoop2/lib/hbase-protocol-0.96.0-hadoop2.jar,

 * /home/q/hbase/hbase-0.96.0-hadoop2/lib/hbase-server-0.96.0-hadoop2.jar

 * ./spark_2.10-1.0.jar

 */

public class SparkFromHbase implements Serializable {

    /**

     * copy from org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil

     *

     * @param scan

     * @return

     * @throws IOException

     */

    String convertScanToString(Scan scan) throws IOException {

        ClientProtos.Scan proto = ProtobufUtil.toScan(scan);

        return Base64.encodeBytes(proto.toByteArray());

    }

    public void start() {

        SparkConf sparkConf = new SparkConf();

        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        Configuration conf = HBaseConfiguration.create();

        Scan scan = new Scan();

        //scan.setStartRow(Bytes.toBytes("195861-1035177490"));

        //scan.setStopRow(Bytes.toBytes("195861-1072173147"));

        scan.addFamily(Bytes.toBytes("cf"));

        scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("col_1"));

        try {

            String tableName = "wyp";

            conf.set(TableInputFormat.INPUT_TABLE, tableName);

            conf.set(TableInputFormat.SCAN, convertScanToString(scan));

            JavaPairRDD<ImmutableBytesWritable, Result> hBaseRDD = sc.newAPIHadoopRDD(conf,

                    TableInputFormat.class, ImmutableBytesWritable.class,

                    Result.class);

            JavaPairRDD<String, Integer> levels = hBaseRDD.mapToPair(

                    new PairFunction<Tuple2<ImmutableBytesWritable, Result>, String, Integer>() {

                        @Override

                        public Tuple2<String, Integer> call(Tuple2<ImmutableBytesWritable, Result> immutableBytesWritableResultTuple2) throws Exception {

                            byte[] o = immutableBytesWritableResultTuple2._2().getValue(Bytes.toBytes("cf"), Bytes.toBytes("col_1"));

                            if (o != null) {

                                return new Tuple2<String, Integer>(new String(o), 1);

                            }

                            return null;

                        }

                    });

            JavaPairRDD<String, Integer> counts = levels.reduceByKey(

                    new Function2<Integer, Integer, Integer>() {

                        @Override

                        public Integer call(Integer i1, Integer i2) {

                            return i1 + i2;

                        }

                    });

            List<Tuple2<String, Integer>> output = counts.collect();

            for (Tuple2 tuple : output) {

                System.out.println(tuple._1() + ": " + tuple._2());

            }

            sc.stop();

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

    public static void main(String[] args) throws InterruptedException {

        new SparkFromHbase().start();

        System.exit(0);

    }

}

这样本段代码段是从Hbase表名为flight_wap_order_log的数据库中读取cf列簇上的airName一列的数据，这样我们就可以对myRDD进行相应的操作：

System.out.println(myRDD.count());

本段代码需要在pom.xml文件加入以下依赖：

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.10</artifactId>
        <version>0.9.1</version>
</dependency>
 
<dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase</artifactId>
        <version>0.98.2-hadoop2</version>
</dependency>
 
<dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-client</artifactId>
        <version>0.98.2-hadoop2</version>
</dependency>
 
<dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-common</artifactId>
        <version>0.98.2-hadoop2</version>
</dependency>
 
<dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-server</artifactId>
        <version>0.98.2-hadoop2</version>
</dependency>

Scala版如下：

import org.apache.spark._
import org.apache.spark.rdd.NewHadoopRDD
import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor}
import org.apache.hadoop.hbase.client.HBaseAdmin
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
 
/////////////////////////////////////////////////////////////////////
 User: 过往记忆
 Date: 14-6-29
 Time: 23:59
 bolg:
 本文地址：/archives/1051
 过往记忆博客，专注于hadoop、hive、spark、shark、flume的技术博客，大量的干货
 过往记忆博客微信公共帐号：iteblog_hadoop
/////////////////////////////////////////////////////////////////////
 
object HBaseTest {
  def main(args: Array[String]) {
    val sc = new SparkContext(args(0), "HBaseTest",
      System.getenv("SPARK_HOME"), SparkContext.jarOfClass(this.getClass))
 
    val conf = HBaseConfiguration.create()
    conf.set(TableInputFormat.INPUT_TABLE, args(1))
 
    val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat],
      classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
      classOf[org.apache.hadoop.hbase.client.Result])
 
    hBaseRDD.count()
 
    System.exit(0)
  }
}

我们需要在加入如下依赖：

libraryDependencies ++= Seq(
        "org.apache.spark" % "spark-core_2.10" % "0.9.1",
        "org.apache.hbase" % "hbase" % "0.98.2-hadoop2",
        "org.apache.hbase" % "hbase-client" % "0.98.2-hadoop2",
        "org.apache.hbase" % "hbase-common" % "0.98.2-hadoop2",
        "org.apache.hbase" % "hbase-server" % "0.98.2-hadoop2"
)

　　在测试的时候，需要配置好Hbase、Hadoop环境，否则程序会出现问题，特别是让程序找到Hbase-site.xml配置文件。

package com.iteblog.spark;
   
  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.protobuf.ProtobufUtil;
  import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
  import org.apache.hadoop.hbase.util.Base64;
  import org.apache.hadoop.hbase.util.Bytes;
  import org.apache.spark.SparkConf;
  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.Serializable;
  import scala.Tuple2;
   
  import java.io.IOException;
  import java.util.List;
   
  /**
  * User: iteblog
  * Date: 14-6-27
  * Time: 下午5:18
  *blog: http://www.iteblog.com
  *
  * Usage: bin/spark-submit --master yarn-cluster --class com.iteblog.spark.SparkFromHbase
  * --jars /home/q/hbase/hbase-0.96.0-hadoop2/lib/htrace-core-2.01.jar,
  * /home/q/hbase/hbase-0.96.0-hadoop2/lib/hbase-common-0.96.0-hadoop2.jar,
  * /home/q/hbase/hbase-0.96.0-hadoop2/lib/hbase-client-0.96.0-hadoop2.jar,
  * /home/q/hbase/hbase-0.96.0-hadoop2/lib/hbase-protocol-0.96.0-hadoop2.jar,
  * /home/q/hbase/hbase-0.96.0-hadoop2/lib/hbase-server-0.96.0-hadoop2.jar
  * ./spark_2.10-1.0.jar
  */
  public class SparkFromHbase implements Serializable {
   
  /**
  * copy from org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil
  *
  * @param scan
  * @return
  * @throws IOException
  */
  String convertScanToString(Scan scan) throws IOException {
  ClientProtos.Scan proto = ProtobufUtil.toScan(scan);
  return Base64.encodeBytes(proto.toByteArray());
  }
   
  public void start() {
  SparkConf sparkConf = new SparkConf();
  JavaSparkContext sc = new JavaSparkContext(sparkConf);
   
   
  Configuration conf = HBaseConfiguration.create();
   
  Scan scan = new Scan();
  //scan.setStartRow(Bytes.toBytes("195861-1035177490"));
  //scan.setStopRow(Bytes.toBytes("195861-1072173147"));
  scan.addFamily(Bytes.toBytes("cf"));
  scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("col_1"));
   
  try {
   
  String tableName = "wyp";
  conf.set(TableInputFormat.INPUT_TABLE, tableName);
  conf.set(TableInputFormat.SCAN, convertScanToString(scan));
   
   
  JavaPairRDD<ImmutableBytesWritable, Result> hBaseRDD = sc.newAPIHadoopRDD(conf,
  TableInputFormat.class, ImmutableBytesWritable.class,
  Result.class);
   
  JavaPairRDD<String, Integer> levels = hBaseRDD.mapToPair(
  new PairFunction<Tuple2<ImmutableBytesWritable, Result>, String, Integer>() {
  @Override
  public Tuple2<String, Integer> call(Tuple2<ImmutableBytesWritable, Result> immutableBytesWritableResultTuple2) throws Exception {
  byte[] o = immutableBytesWritableResultTuple2._2().getValue(Bytes.toBytes("cf"), Bytes.toBytes("col_1"));
  if (o != null) {
  return new Tuple2<String, Integer>(new String(o), 1);
  }
  return null;
  }
  });
   
  JavaPairRDD<String, Integer> counts = levels.reduceByKey(
  new Function2<Integer, Integer, Integer>() {
  @Override
  public Integer call(Integer i1, Integer i2) {
  return i1 + i2;
  }
  });
   
  List<Tuple2<String, Integer>> output = counts.collect();
  for (Tuple2 tuple : output) {
  System.out.println(tuple._1() + ": " + tuple._2());
  }
   
  sc.stop();
   
  } catch (Exception e) {
  e.printStackTrace();
  }
  }
   
  public static void main(String[] args) throws InterruptedException {
  new SparkFromHbase().start();
  System.exit(0);
  }
  }

最新文章

  1. SeaJS 模块化加载框架使用
  2. SQL Server 执行计划中的扫描方式举例说明
  3. 【转】linux下mkisofs制作光盘映像cdrecord刻录光盘
  4. Objective-C运行时编程 - 实现自动化description方法的思路及代码示例
  5. 实现一个不停发包的Android应用（类似于电脑上的Ping命令）
  6. 不需要软件让Windows7变身WIFI热点
  7. python字符串-内置方法列举
  8. react事件处理函数中绑定this的bind()函数
  9. nginx 学习（一）
  10. Django适合做大用户量的系统吗？
  11. JS通过键盘点击事件实现div移动
  12. 《开始使用Linux》单元测验 1
  13. 2.1Python基础语法（一）之注释与数据类型：
  14. libnids
  15. com.panie 项目开发随笔_数据字典（2017.2.24）
  16. c++官方文档-命名空间
  17. 2基本类型数组和枚举类型——重拾Java
  18. PropertySheet
  19. Deep learning for visual understanding: A review
  20. BUG-jQuery提交表单submit方法-TypeError: e[h] is not a function

热门文章

  1. Gym101612H Hidden Supervisors
  2. hdu2093 考试排名（还需完善）
  3. vim的基础命令
  4. SQL面试50题
  5. eatwhatApp开发实战（八）
  6. [PHP自动化－进阶]005.Snoopy采集框架介绍
  7. DC-8靶机writeup
  8. 16 . PythonWeb框架之Django
  9. 《学习scrapy框架爬小说》的进一步完善
  10. Alpha冲刺——4.30