1, hbase测试环境搭建

https://www.yuque.com/tiankonghewo/note/mlemax

2. saveAsNewAPIHadoopDataset

1, row 映射到 put

  1. import java.text.SimpleDateFormat
  2. import java.util.Date
  3. import org.apache.hadoop.hbase.client.Put
  4. import org.apache.spark.sql.Row
  6. object RowMapper {
  7.     def rowMapper(row: Row): Put = {
  8.         import org.apache.hadoop.hbase.util.Bytes
  9.         val column0_id = row.getLong(0)
  10.         val column1_name = row.getDecimal(1)
  11.         val end_time = new SimpleDateFormat("yyyyMMdd").format(new Date())
  12.         val put = new Put(Bytes.toBytes(end_time + "-" + column0_customer_id.toString))
  13.         put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("id"), Bytes.toBytes(column0_id.toString))
  14.         put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("name"), Bytes.toBytes(column1_name.toString))
  15.         put
  16.     }
  17. }

2.df 写入 hbase

  1. import java.io.IOException
  2. import java.util.Properties
  3. import org.apache.hadoop.hbase.client.{Connection, HTable, Put}
  4. import org.apache.spark.sql.{DataFrame, Row}
  6. object HbaseWriter {
  7.   //批量写hbase，效率更高
  8.   def hbasebulk(df: DataFrame, prop: Properties, rowMapper: Row => Put): Unit = {
  9.     import org.apache.hadoop.hbase.client.{Put, Result}
  10.     import org.apache.hadoop.hbase.io.ImmutableBytesWritable
  11.     import org.apache.hadoop.hbase.mapreduce.TableOutputFormat
  12.     import org.apache.hadoop.mapreduce.Job
  13.     val hbaseConf = org.apache.hadoop.hbase.HBaseConfiguration.create()
  14.     hbaseConf.set("hbase.zookeeper.quorum", prop.getProperty("hbase.zookeeper.quorum"))
  15.     hbaseConf.set("hbase.zookeeper.property.clientPort", prop.getProperty("hbase.zookeeper.property.clientPort"))
  16.     hbaseConf.set(TableOutputFormat.OUTPUT_TABLE, prop.getProperty("hbase_table_name"))
  17.     val job = Job.getInstance(hbaseConf)
  18.     job.setOutputKeyClass(classOf[ImmutableBytesWritable])
  19.     job.setOutputValueClass(classOf[Result])
  20.     job.setOutputFormatClass(classOf[TableOutputFormat[ImmutableBytesWritable]])
  21.     val rdd = df.rdd.map(row => (new ImmutableBytesWritable, rowMapper(row)))
  22.     rdd.saveAsNewAPIHadoopDataset(job.getConfiguration)
  23.   }
  24. }

3, hortonworks hbase connector

1, dependency

  1.         <dependency>
  2.             <groupId>io.netty</groupId>
  3.             <artifactId>netty-all</artifactId>
  4.             <version>4.1.18.Final</version>
  5.         </dependency>
  6.         <dependency>
  7.             <groupId>com.hortonworks</groupId>
  8.             <artifactId>shc-core</artifactId>
  9.             <version>1.1.1-2.1-s_2.11</version>
  10.         </dependency>

2, code

  1. package com.wsy.learn
  3. import org.apache.spark.sql.execution.datasources.hbase.HBaseTableCatalog
  4. import org.apache.spark.sql.{SaveMode, SparkSession}
  6. object HbaseHortonworks {
  8.     //此case class 不能定义在方法中,否则运行报错
  9.     case class Student(rowkey: String, spark_height: String, spark_birthday: String)
  11.     def main(args: Array[String]): Unit = {
  12.         val spark = SparkSession
  13.                 .builder()
  14.                 .master("local")
  15.                 .appName(this.getClass.getSimpleName)
  16.                 .getOrCreate()
  18.         val catlog =
  19.             s"""{
  20.                |"table":{"namespace":"default", "name":"student", "tableCoder":"PrimitiveType"},
  21.                |"rowkey":"rowkey",
  22.                |"columns":{
  23.                |"rowkey":{"cf":"rowkey", "col":"rowkey", "type":"string"},
  24.                |"spark_height":{"cf":"cf", "col":"height", "type":"string"},
  25.                |"spark_birthday":{"cf":"cf", "col":"birthday", "type":"string"}
  26.                |}
  27.                |}""".stripMargin
  29.         var newData = Seq(Student("wty", "62", "20141001"), Student("son", "22", "20210110"))
  30.         val rdd = spark.sparkContext.parallelize(newData)
  31.         import spark.implicits._
  32.         val df1 = rdd.toDF
  33.         df1.show()
  34.         df1.write.mode(SaveMode.Append)
  35.                 .option("hbase.zookeeper.quorum", "myhbase:2181")
  36.                 .options(Map(HBaseTableCatalog.tableCatalog -> catalog, HBaseTableCatalog.newTable -> "5"))
  37.                 .format("org.apache.spark.sql.execution.datasources.hbase")
  38.                 .save()
  40.         val df = spark
  41.                 .read
  42.                 .option("hbase.zookeeper.quorum", "myhbase:2181")
  43.                 .options(Map(HBaseTableCatalog.tableCatalog -> catalog))
  44.                 .format("org.apache.spark.sql.execution.datasources.hbase")
  45.                 .load()
  46.         df.show()
  48.         spark.stop()
  49.     }
  50. }

hortonworks-spark官网
https://github.com/hortonworks-spark/shc

  1. Given a DataFrame with specified schema, above will create an HBase table with 5 regions and save the DataFrame inside. 
  2. Note that if HBaseTableCatalog.newTable is not specified, the table has to be pre-created.

https://hbase.apache.org/book.html#_sparksqldataframes
https://docs.microsoft.com/en-us/azure/hdinsight/hdinsight-using-spark-query-hbase
case class 定义位置bug
https://blog.csdn.net/zhao897426182/article/details/78284327
https://stackoverflow.com/questions/33704831/value-todf-is-not-a-member-of-org-apache-spark-rdd-rdd

4, alibaba hbase connector

1, dependency

  1.         <dependency>
  2.             <groupId>com.aliyun.apsaradb</groupId>
  3.             <artifactId>alihbase-spark</artifactId>
  4.             <version>1.1.3_2.3.2-1.0.3</version>
  5.         </dependency>

2, 官方代码

  1. import org.apache.spark.sql.SparkSession
  3. object Hello {
  4.     def main(args: Array[String]): Unit = {
  5.         val spark = SparkSession
  6.                 .builder()
  7.                 .master("local")
  8.                 .appName(this.getClass.getSimpleName)
  9.                 .getOrCreate()
  11.         val sparkTableName = "x"
  12.         val hbaseTableName = "student"
  13.         val zkAddress = "myhbase:2181"
  14.         val createCmd =
  15.             s"""CREATE TABLE ${sparkTableName} USING org.apache.hadoop.hbase.spark
  16.                |    OPTIONS ('catalog'=
  17.                |    '{"table":{"namespace":"default", "name":"${hbaseTableName}"},"rowkey":"rowkey",
  18.                |    "columns":{
  19.                |    "rowkey":{"cf":"rowkey", "col":"rowkey", "type":"string"},
  20.                |    "spark_height":{"cf":"cf", "col":"height", "type":"string"},
  21.                |    "spark_birthday":{"cf":"cf", "col":"birthday", "type":"String"}}}',
  22.                |    'hbase.zookeeper.quorum' = '${zkAddress}'
  23.                |    )""".stripMargin
  25.         spark.sql(createCmd)
  26.         spark.sql("select * from " + sparkTableName + " limit 5").show
  27.         spark.stop()
  28.     }
  29. }

3, spark.read

  1. package com.wsy.learn
  3. import org.apache.spark.sql.datasources.hbase.HBaseTableCatalog
  4. import org.apache.spark.sql.{SaveMode, SparkSession}
  6. object HbaseAlibaba {
  8.     case class Student(rowkey: String, spark_height: String, spark_birthday: String)
  10.     def main(args: Array[String]): Unit = {
  11.         val spark = SparkSession
  12.                 .builder()
  13.                 .master("local")
  14.                 .appName(this.getClass.getSimpleName)
  15.                 .getOrCreate()
  17.         val catalog =
  18.             """
  19.               |{
  20.               |    "table": {
  21.               |        "namespace": "default",
  22.               |        "name": "student"
  23.               |    },
  24.               |    "rowkey": "key",
  25.               |    "columns": {
  26.               |        "rowkey": {
  27.               |            "cf": "rowkey",
  28.               |            "col": "key",
  29.               |            "type": "string"
  30.               |        },
  31.               |        "spark_height": {
  32.               |            "cf": "cf",
  33.               |            "col": "height",
  34.               |            "type": "string"
  35.               |        },
  36.               |        "spark_birthday": {
  37.               |            "cf": "cf",
  38.               |            "col": "birthday",
  39.               |            "type": "string"
  40.               |        }
  41.               |    }
  42.               |}
  43.             """.stripMargin
  45.         var newData = Seq(Student("wty", "61", "20141001"), Student("son", "21", "20210110"))
  46.         val rdd = spark.sparkContext.parallelize(newData)
  47.         import spark.implicits._
  48.         val df1 = rdd.toDF()
  49.         //todo 读取是没问题的,写入只能新建表,是个bug,有待
  50.         df1.write.mode(SaveMode.Append)
  51.                 .option("hbase.zookeeper.quorum", "myhbase:2181")
  52.                 .options(Map(HBaseTableCatalog.tableCatalog -> catalog, HBaseTableCatalog.newTable -> "2"))
  53.                 .format("org.apache.hadoop.hbase.spark")
  54.                 .save()
  56.         val df = spark
  57.                 .read
  58.                 .option("hbase.zookeeper.quorum", "myhbase:2181")
  59.                 .options(Map(HBaseTableCatalog.tableCatalog -> catalog))
  60.                 .format("org.apache.hadoop.hbase.spark")
  61.                 .load()
  62.         df.show()
  64.         spark.stop()
  65.     }
  66. }

https://help.aliyun.com/document_detail/93908.html?spm=a2c6h.12873639.0.0.f45f9079aM2gH8
https://github.com/aliyun/aliyun-apsaradb-hbase-demo/blob/master/spark/spark-examples/src/main/scala/com/aliyun/spark/hbase/SparkOnHBaseSparkSession.scala