大数据面试之Spark SQL

谁践踏了优雅 2023-02-17 15:54 105阅读 0赞

### 大数据面试之Spark SQL ###

*  1.Spark SQL
 *   *  1.1 Spark SQL连接MySql
     *  1.2 Spark读取Hive数据
     *  1.3 DataFrame 和 RDD 之间的转换？
     *  1.4 Spark读取HBase数据
     *  1.5 Spark写入HBase数据

说明，感谢亮哥长期对我的帮助，此处多篇文章均为亮哥带我整理。以及参考诸多博主的文章。如果侵权，请及时指出，我会立马停止该行为；如有不足之处，还请大佬不吝指教，以期共同进步。

# 1.Spark SQL #

## 1.1 Spark SQL连接MySql ##

val conf = new SparkConf().setAppName("TestMysql").setMaster("local")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    
    val url = "jdbc:mysql://192.168.123.102:3306/hivedb"
    val table = "dbs"
    val properties = new Properties()
    properties.setProperty("user","root")
    properties.setProperty("password","root")
    //需要传入Mysql的URL、表明、properties（连接数据库的用户名密码）
    val df = sqlContext.read.jdbc(url,table,properties)
    df.createOrReplaceTempView("dbs")
    sqlContext.sql("select * from dbs").show()

## 1.2 Spark读取Hive数据 ##

开发环境则把resource文件夹下添加hive-site.xml文件
    集群环境把hive的配置文件要发到$SPARK_HOME/conf目录下
    
    val conf = new SparkConf().setMaster("local").setAppName(this.getClass.getSimpleName)
    val sc = new SparkContext(conf)
    val sqlContext = new HiveContext(sc)
    sqlContext.sql("select * from myhive.student").show()

[参考][Link 1]

## 1.3 DataFrame 和 RDD 之间的转换？ ##

1.  通过 case class 创建 DataFrames（反射）

// 将本地的数据读入 RDD， 并将 RDD 与 case class 关联
    val peopleRDD = sc.textFile("E:\\666\\people.txt")
      .map(line => People(line.split(",")(0), line.split(",")(1).trim.toInt))
    import context.implicits._
    // 将RDD 转换成 DataFrames
    val df = peopleRDD.toDF
    //将DataFrames创建成一个临时的视图
    df.createOrReplaceTempView("people")
    //使用SQL语句进行查询
    context.sql("select * from people").show()

1.  通过 structType 创建 DataFrames（编程接口）

val fileRDD = sc.textFile("E:\\666\\people.txt")
    // 将 RDD 数据映射成 Row，需要 import org.apache.spark.sql.Row
    val rowRDD: RDD[Row] = fileRDD.map(line => {
      val fields = line.split(",")
      Row(fields(0), fields(1).trim.toInt)
    })
    // 创建 StructType 来定义结构
    val structType: StructType = StructType(
      //字段名，字段类型，是否可以为空
      StructField("name", StringType, true) ::
        StructField("age", IntegerType, true) :: Nil
    )
    /**
      * rows: java.util.List[Row],
      * schema: StructType
      * */
    val df: DataFrame = sqlContext.createDataFrame(rowRDD,structType)
    df.createOrReplaceTempView("people")
    sqlContext.sql("select * from people").show()

1.  通过 json 文件创建 DataFrames

val sqlContext = new SQLContext(sc)
    val df: DataFrame = sqlContext.read.json("E:\\666\\people.json")
    df.createOrReplaceTempView("people")
    sqlContext.sql("select * from people").show()

[参考][Link 1]

## 1.4 Spark读取HBase数据 ##

val zookeeper_ip="192.168.199.128,192.168.199.131,192.168.199.132"
    val tableName="user"
    
    val spark= SparkSession.builder()
      .appName("readHbase_dataframe")
      .master("local")
      .getOrCreate()
    
    val hbaseConf = HBaseConfiguration.create()
    hbaseConf.set("hbase.zookeeper.quorum", zookeeper_ip)
    hbaseConf.set("hbase.zookeeper.property.clientPort", "2181")
    hbaseConf.set(TableInputFormat.INPUT_TABLE, tableName)
    
    val sqlContext = spark.sqlContext
    val sc=spark.sparkContext
    import sqlContext.implicits._ //没有的话 rdd不能调用.toDF
    // 从数据源获取数据
    val hbaseRDD = sc.newAPIHadoopRDD(hbaseConf,classOf[TableInputFormat],classOf[ImmutableBytesWritable],classOf[Result])
    
    // 将数据映射为表  也就是将 RDD转化为 dataframe schema
    val hbaseTable=hbaseRDD.map(r=>{
      val hbase_result=r._2
      val rowkey=Bytes.toString(hbase_result.getRow)
      val name=Bytes.toString(hbase_result.getValue(Bytes.toBytes("info"),Bytes.toBytes("name")))
      val age=Bytes.toString(hbase_result.getValue(Bytes.toBytes("info"),Bytes.toBytes("age")))
    
      (rowkey,name,age)
    }).toDF("id","name","age")
    
    hbaseTable.createOrReplaceTempView("emp")
    val hbase_result=spark.sql("select * from emp")
    hbase_result.show()

## 1.5 Spark写入HBase数据 ##

val zookeeper_ip="172.24.112.13,172.24.112.14,172.24.112.15"
    val tableName="testTable"
    
    System.setProperty("HADOOP_USER_NAME", "hdfs")
    
    val spark= SparkSession.builder()
      .appName("ReadHbase_Dataframe")
      .master("local")
      .getOrCreate()
    
    val hbaseConf = HBaseConfiguration.create()
    hbaseConf.set("hbase.zookeeper.quorum", zookeeper_ip)
    hbaseConf.set("hbase.zookeeper.property.clientPort", "2181")
    hbaseConf.set(TableInputFormat.INPUT_TABLE, tableName)
    
    val hbase_list=HbaseUtils.getAllData(tableName)//用到hbase，hdfs的简单工具类
    if(hdfsUtil.isExists("hdfs://172.24.112.11:8020/input/hbase.csv")){
      hdfsUtil.deleteHDFSFile("hdfs://172.24.112.11:8020/input/hbase.csv")
    }
    for(context<-hbase_list){
      println(context)
      hdfsUtil.append("hdfs://172.24.112.11:8020/input/hbase.csv",context+"\n")
    }
    
    val myschema = StructType(List(StructField("rowkey", DataTypes.StringType)
      , StructField("family", DataTypes.StringType)
      ,StructField("column", DataTypes.StringType)
      ,StructField("value", DataTypes.StringType)
      ))
    val df=spark.read.schema(myschema).csv("hdfs://172.24.112.11:8020/input/hbase.csv")
    df.createOrReplaceTempView("emp")
    val hbase_result=spark.sql("select * from emp")
    hbase_result.show()

[Link 1]: https://www.cnblogs.com/qingyunzong/p/8987579.html