Spark 读 Hbase

package com.grady

import org.apache.hadoop.hbase.HBaseConfiguration

import org.apache.hadoop.hbase.client.Result

import org.apache.hadoop.hbase.io.ImmutableBytesWritable

import org.apache.hadoop.hbase.mapreduce.TableInputFormat

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

import org.apache.log4j.{Level, Logger}

import org.apache.spark.sql.SparkSession

object SparkReadHbase {

  def main(args: Array[String]): Unit = {

    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)

//    val sparkConf = new SparkConf().setAppName("SparkReadHbase")

//    val sc = new SparkContext(sparkConf)

    // 也可以通过sparkSession 来后去SparkContext

    val spark = SparkSession.builder().appName("SparkReadHbase").getOrCreate()

    val sc = spark.sparkContext

    val tablename = "jiang:student"

    val hbaseConf = HBaseConfiguration.create()

    hbaseConf.set("hbase.zookeeper.quorum","10.82.232.64")

    hbaseConf.set("hbase.zookeeper.property.clientPort", "2181")

    hbaseConf.set("zookeeper.znode.parent", "/hbase")

    //注意: 这里是INPUT_TABLE, 写是OUTPUT_TABLE

    hbaseConf.set(TableInputFormat.INPUT_TABLE, tablename)

    val hbaseRDD = sc.newAPIHadoopRDD(hbaseConf,

      //TableInputFormat 是 org.apache.hadoop.hbase.mapreduce 包下的

      classOf[TableInputFormat],

      classOf[ImmutableBytesWritable],

      classOf[Result])

    hbaseRDD.foreach{

      case (_, result) =>

        //获取行键

        val key = Bytes.toString(result.getRow)

        //通过列族和列名获取列

        val name = Bytes.toString(result.getValue("cf".getBytes, "name".getBytes))

        val age = Bytes.toString(result.getValue("cf".getBytes, "age".getBytes))

        println("Row key:"+key+"\tcf1.Name:"+name+"\tcf1.Age:"+age)

    }

    spark.stop()

  }

}

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

    <parent>

        <artifactId>spark-practise</artifactId>

        <groupId>org.example</groupId>

        <version>1.0-SNAPSHOT</version>

    </parent>

    <modelVersion>4.0.0</modelVersion>

    <artifactId>usehive1</artifactId>

    <repositories>

        <repository>

            <id>scala-tools.org</id>

            <name>Scala-Tools Maven2 Repository</name>

            <url>http://scala-tools.org/repo-releases</url>

        </repository>

    </repositories>

    <pluginRepositories>

        <pluginRepository>

            <id>scala-tools.org</id>

            <name>Scala-Tools Maven2 Repository</name>

            <url>http://scala-tools.org/repo-releases</url>

        </pluginRepository>

    </pluginRepositories>

    <dependencies>

        <dependency>

            <groupId>org.scala-lang</groupId>

            <artifactId>scala-library</artifactId>

            <version>${scala.version}</version>

        </dependency>

        <dependency>

            <groupId>junit</groupId>

            <artifactId>junit</artifactId>

            <version>4.4</version>

            <scope>test</scope>

        </dependency>

        <dependency>

            <groupId>org.specs</groupId>

            <artifactId>specs</artifactId>

            <version>1.2.5</version>

            <scope>test</scope>

        </dependency>

        <dependency>

            <groupId>org.apache.spark</groupId>

            <artifactId>spark-core_${scala.binary.version}</artifactId>

            <version>${spark.version}</version>

        </dependency>

        <dependency>

            <groupId>org.apache.spark</groupId>

            <artifactId>spark-sql_${scala.binary.version}</artifactId>

            <version>${spark.version}</version>

        </dependency>

        <dependency>

            <groupId>org.apache.hbase</groupId>

            <artifactId>hbase-client</artifactId>

        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.hbase.connectors.spark/hbase-spark -->

        <dependency>

            <groupId>org.apache.hbase.connectors.spark</groupId>

            <artifactId>hbase-spark</artifactId>

        </dependency>

    </dependencies>

    <build>

        <sourceDirectory>src/main/scala</sourceDirectory>

        <testSourceDirectory>src/test/scala</testSourceDirectory>

        <plugins>

            <plugin>

                <groupId>org.scala-tools</groupId>

                <artifactId>maven-scala-plugin</artifactId>

                <executions>

                    <execution>

                        <goals>

                            <goal>compile</goal>

                            <goal>testCompile</goal>

                        </goals>

                    </execution>

                </executions>

                <configuration>

                    <scalaVersion>${scala.version}</scalaVersion>

                    <args>

                        <arg>-target:jvm-1.8</arg>

                    </args>

                </configuration>

            </plugin>

        </plugins>

    </build>

</project>

执行：spark-submit --master local[2] --num-executors 10 --class com.grady.SparkReadHbase /app/data/appdeploy/usehive1-1.0-SNAPSHOT.jar

日志：

Row key:1       cf1.Name:jack   cf1.Age:15

Row key:2       cf1.Name:Lily   cf1.Age:16

Row key:3       cf1.Name:mike   cf1.Age:16

巴特西

Spark 读 Hbase

最新文章

热门文章