描述一下需求

有個表的結構如下

    Emp (
           Eno CHAR(4)，
           Ename CHAR(8)，
           Esex CHAR(1) CHECK(Esex IN  ('M','F'))，
           EDno CHAR(4) REFERENCES Dept (Dno)，
           PRIMARY KEY (Eno)
    )；
    Dept (
           Dno CHAR(4)  NOT NULL UNIQUE，
           Dname CHAR(20)，
           Daddr CHAR(30)
    )；
 

上表簡單的分為Emp.txt和Dept.txt兩個文字檔案，資料具體內容自定義

用Spark和Spark SQL分別實現以下功能：

1.提取Eno，Ename，Esex，Dname，Daddr

2.用Avro格式將資料存到指定目錄下

在這裡主要想記錄的是:

1.使用spark,sql如何進行表的連線（Join）

2.如何將資料儲存為需要的格式

先放pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<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/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.huangxiao</groupId>
    <artifactId>local_spark</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.3.0</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.11</artifactId>
        <version>2.3.0</version>
    </dependency>
        <!-- https://mvnrepository.com/artifact/com.databricks/spark-avro -->
    <dependency>
        <groupId>com.databricks</groupId>
        <artifactId>spark-avro_2.11</artifactId>
        <version>4.0.0</version>
    </dependency>
    </dependencies>

</project>
 

然後spark部分

package scala

import org.apache.spark.sql.{Row, SparkSession, types}
import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}
import com.databricks.spark.avro._

//用spark完成對資料的提取和儲存操作
//用Spark和Spark SQL分別實現以下功能：提取Eno，Ename，Esex，Dname，Daddr，用Avro格式將資料存到xx目錄下

object quiz1_for_spark_521 {
  def main(args: Array[String]) {
    if (args.length < 2) {
      println("Usage:SparkWordCount FileName")
      System.exit(1)
    }

    val conf = new SparkConf().setAppName("log_deal").setMaster("local")
    val sc = new SparkContext(conf)
    val sparkSession = SparkSession.builder().appName("RDD to DataFrame").config(conf).getOrCreate()//新的sql api，等價於sqlContext().


    val Deptfile = sc.textFile(args(1))
    val DeptRDD=Deptfile.map{l =>
      val line = l.split(',')
      val dno=line(0)
      val dname=line(1)
      val daddr=line(2)
      (dno,dname+','+daddr)
    }

    val Empfile = sc.textFile(args(0))
    val EmpRDD=Empfile.map{l =>
      val line=l.split(',')
      val name=line(1)
      val sex=line(2)
      val edno=line(3)
      (edno,line(0)+','+name+','+sex)}
    //join兩個RDD
    val joinRDD=EmpRDD.join(DeptRDD).map(x=>x._2._1+','+x._2._2)
      .map{l=>
      val l_l=l.split(',')
      val eno=l_l(0)
      val ename=l_l(1)
      val esex= l_l(2)
      val dname= l_l(3)
      val daddr= l_l(4)
      Row(eno,ename,esex,dname,daddr)}

    //在這裡需要宣告一下表的結構。因為arvo需要列名
    val schema_final = StructType(
      Seq(
        //這裡的StringType注意別用錯了，是org.apache.spark.sql.types.StringType。不注意的話會用到org.apache.spark.sql.StringType
        StructField("Eno",types.StringType,true)
        ,StructField("Ename",types.StringType,true)
        ,StructField("Esex",types.StringType,true)
        ,StructField("Dname",types.StringType,true)
        ,StructField("Daddr",types.StringType,true)
      )
    )
    //將RDD與表結構進行匹配，生成DF便於avro格式的儲存
    val df=sparkSession.createDataFrame(joinRDD,schema_final)
    //這個方法是cloudera官網的例項，很多用法都有具體的例子 https://www.cloudera.com/documentation/enterprise/latest/topics/spark_avro.html  
    df.write.format("com.databricks.spark.avro").save(args(2))
    sc.stop()
  }

}
 

sparkSQL部分

package scala

import org.apache.spark.sql.{Row, SparkSession, types}
import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}
import com.databricks.spark.avro._

object quiz1_for_sql_521 {
  def main(args: Array[String]) {
    if (args.length < 2) {
      println("Usage:SparkWordCount FileName")
      System.exit(1)
    }

    val conf = new SparkConf().setAppName("log_deal").setMaster("local")
    val sc = new SparkContext(conf)
    val sparkSession = SparkSession.builder().appName("RDD to DataFrame")
      .config(conf).getOrCreate()

    val EmpFile = sc.textFile(args(0))
    val EmpRDD = EmpFile.map{l=>
      val l_l=l.split(',')
      val eno=l_l(0)
      val ename=l_l(1)
      val esex= l_l(2)
      val edno = l_l(3)
      Row(eno,ename,esex,edno)}

    val schema_E = StructType(
      Seq(
        StructField("Eno",types.StringType,true)
        ,StructField("Ename",types.StringType,true)
        ,StructField("Esex",types.StringType,true)
        ,StructField("Dno",types.StringType,true)
      )
    )
    val df_E=sparkSession.createDataFrame(EmpRDD,schema_E)
//    df_E.createTempView("E")

    val DeptFile = sc.textFile(args(1))
    val DeptRDD = DeptFile.map{l=>
      val l_l=l.split(',')
      val dno=l_l(0)
      val dname=l_l(1)
      val daddr= l_l(2)
      Row(dno,dname,daddr)}

    val schema = StructType(
      Seq(
        StructField("Dno",types.StringType,true)
        ,StructField("Dname",types.StringType,true)
        ,StructField("Daddr",types.StringType,true)
      )
    )


    val df_D=sparkSession.createDataFrame(DeptRDD,schema)
//    df_D.createTempView("D")
    //此處是DataFrame的join方法
    val final_df=df_E.join(df_D,"Dno")

    //這裡用的也是sql新的api,建立一個檢視後就可以中sql語句進行分析啦
    final_df.createTempView("E_D")
    val results = sparkSession.sql("SELECT Eno,Ename,Esex,Dname,Daddr FROM E_D")
    results.show()
    results.write.format("com.databricks.spark.avro").save(args(2))
    sc.stop()
  }

}