Ubuntu下搭建Hbase单机版并贯彻Java API访问

SparkSQL和DataFrame

SparkSQL简介

Spark(Spark)SQL是斯帕克(Spark)用来处理结构化数据的一个模块,它提供了一个编程抽象为做DataFrame并且作为分布式SQL查询引擎的用意。它是以斯帕克(Spark)(Spark)SQL转换成RDD,然后提交至集群执行,执行功能特别抢!

SparkSQL的特性

1.易整合
2.合之数据看情势
3.兼容Hive
4.正式的数连接

DataFrames简介
以及RDD类似,DataFrame也是一个分布式数据容器。不过DataFrame更像风数据库的第二维表格,除了数量之外,还记下数据的构造音信,即schema。同时,与Hive类似,DataFrame也帮忙嵌套数据类型(struct、array和map)。从API易用性的角度达
看,DataFrame API提供的凡均等拟高层的涉嫌操作,比函数式的RDD
API要更加自己,门槛再小。由于与R和Pandas的DataFrame类似,斯帕克(Spark)DataFrame很好地延续了风单机数据解析的支出体验。

创建DataFrames

于斯帕克(Spark)SQL中SQLContext是开创DataFrames和执行SQL的进口,在spark-1.5.2中已停放了一个sqlContext
1.在该地创设一个文书,有三列,分别是id、name、age,用空格分隔,然后上传来hdfs上
hdfs dfs -put person.txt /
2.每当spark shell执行下发号施令,读取数据,将各类一行的多少应用列分隔符分割
val lineRDD = sc.textFile(“hdfs://node1:9000/person.txt”).map(_.split(”
“))
3.定义case class(相当给表底schema)
case class Person(id:Int, name:String, age:Int)
4.将RDD和case class关联
val personRDD = lineRDD.map(x => Person(x(0).toInt, x(1),
x(2).toInt))
5.将RDD转换成DataFrame
val personDF = personRDD.toDF
6.对准DataFrame举办处理
personDF.show

DataFrames常见操作

1.//查看DataFrame中之始末
personDF.show
2.//查看DataFrame部分列着之情节
personDF.select(personDF.col(“name”)).show
personDF.select(col(“name”), col(“age”)).show
personDF.select(“name”).show
3.//打印DataFrame的Schema信息
personDF.printSchema
4.//查询所有的name和age,并拿age+1
personDF.select(col(“id”), col(“name”), col(“age”) + 1).show
personDF.select(personDF(“id”), personDF(“name”), personDF(“age”) +
1).show
5.//过滤age大于等于18底
personDF.filter(col(“age”) >= 18).show
6.//按岁举行分组并总计相同年龄的人口
personDF.groupBy(“age”).count().show()

使用SparkSQL风格
比方想行使SQL风格的语法,需要用DataFrame注册成表
personDF.registerTempTable(“t_person”)

1.//查询岁数最深之眼前片叫
sqlContext.sql(“select * from t_person order by age desc limit
2”).show
2.//突显表的Schema音信
sqlContext.sql(“desc t_person”).show

sqlContext.sql()中的情跟描绘普通的基本一致,可是倘诺小心SparkSQL不援助子查询

编写程序执行斯帕克(Spark)(Spark)SQL语句
1.首先在maven项目的pom.xml中添加Spark SQL的依赖

<dependency>  
    <groupId>org.apache.spark</groupId>  
    <artifactId>spark-sql_2.10</artifactId>  
    <version>1.5.2</version>  
</dependency>  

2.有血有肉写法1动case class

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext

object InferringSchema {
  def main(args: Array[String]) {

//创建SparkConf()并设置App名称
val conf = new SparkConf().setAppName("SQL-1")
//SQLContext要依赖SparkContext
val sc = new SparkContext(conf)
//创建SQLContext
val sqlContext = new SQLContext(sc)

//从指定的地址创建RDD
val lineRDD = sc.textFile(args(0)).map(_.split(" "))

//创建case class
//将RDD和case class关联
val personRDD = lineRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt))
//导入隐式转换,如果不到人无法将RDD转换成DataFrame
//将RDD转换成DataFrame
import sqlContext.implicits._
val personDF = personRDD.toDF
 //注册表
    personDF.registerTempTable("t_person")
    //传入SQL
    val df = sqlContext.sql("select * from t_person order by age desc limit 2")
    //将结果以JSON的方式存储到指定位置
    df.write.json(args(1))
    //停止Spark Context
    sc.stop()
  }
}
//case class一定要放到外面
case class Person(id: Int, name: String, age: Int)

以次第打成jar包,上传到spark集群,提交斯帕克(Spark)任务
/usr/local/spark-1.5.2-bin-hadoop2.6/bin/spark-submit  
–class spark.sql.InferringSchema  
–master spark://node1:7077  
/root/spark-mvn-1.0-SNAPSHOT.jar  
hdfs://node1:9000/person.txt  
hdfs://node1:9000/out

查运行结果
hdfs dfs -cat hdfs://node1:9000/out/part-r-*

3.有血有肉写法2,通过StructType间接指定Schema

import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.sql.types._
import org.apache.spark.{SparkContext, SparkConf}

object SpecifyingSchema {
  def main(args: Array[String]) {
    //创建SparkConf()并设置App名称
val conf = new SparkConf().setAppName("SQL-2")
    //SQLContext要依赖SparkContext
    val sc = new SparkContext(conf)
    //创建SQLContext
    val sqlContext = new SQLContext(sc)
    //从指定的地址创建RDD
    val personRDD = sc.textFile(args(0)).map(_.split(" "))
    //通过StructType直接指定每个字段的schema
    val schema = StructType(
      List(
        StructField("id", IntegerType, true),
        StructField("name", StringType, true),
        StructField("age", IntegerType, true)
      )
    )
    //将RDD映射到rowRDD
    val rowRDD = personRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).toInt))
    //将schema信息应用到rowRDD上
    val personDataFrame = sqlContext.createDataFrame(rowRDD, schema)
    //注册表
    personDataFrame.registerTempTable("t_person")
    //执行SQL
    val df = sqlContext.sql("select * from t_person order by age desc limit 4")
    //将结果以JSON的方式存储到指定位置
    df.write.json(args(1))
    //停止Spark Context
    sc.stop()
  }
}

从今MySQL中加载数据(Spark(Spark) Shell模式)
1.启动斯帕克(Spark)(Spark) Shell,必须指定mysql连接驱动jar包
/usr/local/spark-1.5.2-bin-hadoop2.6/bin/spark-shell  
–master spark://node1:7077  
–jars
/usr/local/spark-1.5.2-bin-hadoop2.6/mysql-connector-java-5.1.35-bin.jar
 
–driver-class-path
/usr/local/spark-1.5.2-bin-hadoop2.6/mysql-connector-java-5.1.35-bin.jar

2.打mysql中加载数据
val jdbcDF = sqlContext.read.format(“jdbc”).options(Map(“url” ->
“jdbc:mysql://XXX:3306/bigdata”, “driver” -> “com.mysql.jdbc.Driver”,
“dbtable” -> “person”, “user” -> “root”, “password” ->
“123456”)).load()

3.履行查询
jdbcDF.show()

将数据写入到MySQL中

import java.util.Properties
import org.apache.spark.sql.{SQLContext, Row}
import org.apache.spark.sql.types.{StringType, IntegerType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}

object JdbcRDD {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("MySQL-Demo")
    val sc = new SparkContext(conf)
 val sqlContext = new SQLContext(sc)
    //通过并行化创建RDD
    val personRDD = sc.parallelize(Array("1 tom 5", "2 jerry 3", "3 kitty 6")).map(_.split(" "))
    //通过StructType直接指定每个字段的schema
    val schema = StructType(
      List(
        StructField("id", IntegerType, true),
        StructField("name", StringType, true),
        StructField("age", IntegerType, true)
      )
    )
    //将RDD映射到rowRDD
    val rowRDD = personRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).toInt))
    //将schema信息应用到rowRDD上
    val personDataFrame = sqlContext.createDataFrame(rowRDD, schema)
    //创建Properties存储数据库相关属性
    val prop = new Properties()
    prop.put("user", "root")
    prop.put("password", "123456")
    //将数据追加到数据库
    personDataFrame.write.mode("append").jdbc("jdbc:mysql://192.168.10.1:3306/bigdata", "bigdata.person", prop)
    //停止SparkContext
    sc.stop()
  }
}

hive on spark-SQL

1.安装hive,修改元数据库,加上hive-site.xml(mysql连接)
2.以hive-site.xml文件拷贝到spark集群的conf下
3.将mysql.jar拷贝到spark.lib下
4.执行:sqlCOntext.sql(“select * from table1”).show()
.write.mode(“append”)
.jdbc()
.foreachPartition(it=>{
1.起首化连接
2.it.map(x=>{
描绘多少到存储层
})
3.关连接
})

网上的代码大多冗长,而且无必然能跑通,其实只要促成访问,只待简单几实践代码:

6、导入看重包

8、运行

3、进入Hbase的bin目录启动start-hbase.sh程序

}
catch (IOException e) {
  e.printStackTrace();
}
}

工具:Ubuntu12.04 、Eclipse、Java、Hbase

2、安装Hbase,直接解压.tar 文件即可

4、举办shell操作。执行命令hbase shell,进入shell
。执行同一长长的成立表的操作create ‘table1′,’con1’

}

public class Hello {
public static void main(String[]args)
{
try
{
  Configuration HBASE_CONFIG = new Configuration();
  HBASE_CONFIG.set(“hbase.zookeeper.quorum”, “ubuntu”);
  HBaseConfiguration cfg = new HBaseConfiguration(HBASE_CONFIG);
  HBaseAdmin admin = new HBaseAdmin(cfg);
  if (admin.tableExists(“te”)) {
    System.out.println(“te” + “存在!”);
    System.exit(0);
  }
  else {
    HTableDescriptor tableDesc = new HTableDescriptor(“te”);
    tableDesc.addFamily(new HColumnDescriptor(“co”));
    admin.createTable(tableDesc);
    System.out.println(“成立表成功!”);
  }

7、编写程序

1、在Ubuntu上安装Eclipse,可以在Ubuntu的软件基本直接装,也可透过命令安装,第一差安装失败了,又尝试了平等次,开端产充斥很缓慢,后来赶紧了四起,花费了森日,总算是形成了。需要小心的凡,Eclipse是于带Java的,尽管原先Ubuntu中安了Java,会和夫Java争论,导致Eclipse启动不起。这一个时节要编辑/etc/profile中的Java路径,需要配成Eclipse自带安装的Java路径,一般在/usr/lib/jvm/下

5、新建java项目

import java.io.IOException;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.conf.Configuration;

重点包括commons-logging、hadoop-core、hbase-0.90.4、log4j、zookeeper等六个,这里hbase的教最好及安的hbase版本统一

以此时段系统会卡壳住不动,这时需要修改host,需要以127.0.0.1反化你本机的Ip地址

发表评论

电子邮件地址不会被公开。 必填项已用*标注