Spark读写Hbase中的数据

def main(args: Array[String]) { val sparkConf = new SparkConf().setMaster("local").setAppName("cocapp").set("spark.kryo.registrator", classOf[HBaseConfiguration].getName) .set("spark.executor.memory", "4g") val sc: SparkContext = new SparkContext(sparkConf) val sqlContext = new HiveContext(sc) val mySQLUrl = "jdbc:mysql://localhost:3306/yangsy?user=root&password=yangsiyi" val rows = sqlContext.jdbc(mySQLUrl, "person") val tableName = "spark" val columnFamily = "cf" //rows.first().getString(1) val configuration = HBaseConfiguration.create(); configuration.set(TableInputFormat.INPUT_TABLE, "test"); val admin = new HBaseAdmin(configuration) val hBaseRDD = sc.newAPIHadoopRDD(configuration, classOf[TableInputFormat], classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], classOf[org.apache.hadoop.hbase.client.Result]) hBaseRDD.count()

def toHbase(rows: DataFrame,tableName : String,columnFamily: String) { val configuration = HBaseConfiguration.create(); val admin = new HBaseAdmin(configuration) if (admin.tableExists(tableName)) { print("table Exists") admin.disableTable(tableName); admin.deleteTable(tableName); } configuration.addResource("hbase-site.xml") val tableDesc = new HTableDescriptor(tableName) tableDesc.addFamily(new HColumnDescriptor(columnFamily)) admin.createTable(tableDesc) rows.foreachPartition { row => val table = new HTable(configuration, tableName) row.foreach { a => val put = new Put(Bytes.toBytes("row1")) put.add(Bytes.toBytes(columnFamily), Bytes.toBytes("coulumn1"), Bytes.toBytes(a.getString(0))) table.put(put) println("insert into success") } }