Spark 2.4.0编程指南--spark dataSet action

时间  2019-11-13
标签 spark 2.4.0 编程 指南 dataset action 栏目 Spark 繁體版
原文   原文链接

Spark 2.4.0编程指南--spark dataSet action

更多资源

视频

<iframe width="800" height="500" src="//player.bilibili.com/player.html?aid=38193405&cid=67137841&page=3" scrolling="no" border="0" frameborder="no" framespacing="0" allowfullscreen="true"> </iframe>html

文档

前置条件

  • 已安装好java(选用的是java 1.8.0_191)
  • 已安装好scala(选用的是scala 2.11.121)
  • 已安装好hadoop(选用的是Hadoop 3.1.1)
  • 已安装好spark(选用的是spark 2.4.0)

技能标签

  • Spark session 建立
  • 在Spark 2.0以后，RDD被数据集(Dataset)取代 ，保留RDD旧api
  • 数据集数据集介绍
  • 读取本地文件(txt,json),HDFS文件
  • 对txt格式文件数据遍历(行数据转成对象)
  • 对json格式文件数据遍历(直接转对象)
  • 数据集的action操做
  • collect,collectAsList,count,describe,first,foreach,head,reduce,show,take,takeAsList,toLocalIterator
  • 官网: http://spark.apache.org/docs/2.4.0/sql-getting-started.html

DataSet(数据集)

数据集是分布式数据集合。数据集是Spark 1.6中添加的一个新接口，它提供了RDD的优点（强类型，使用强大的lambda函数的能力）以及Spark SQL优化执行引擎的优势。数据集能够从JVM对象构造，而后使用功能转换（map，flatMap，filter等）进行操做。数据集API在Scala和Java中可用。 Python没有对Dataset API的支持。但因为Python的动态特性，数据集API的许多好处已经可用（即您能够经过名称天然地访问行的字段row.columnName）。 R的状况相似。

BaseSparkSession

  • 公用获得SparkSession的方法
def sparkSession(isLocal:Boolean = false): SparkSession = {

    if(isLocal){
      master = "local"
      val spark = SparkSession.builder
        .master(master)
        .appName(appName)
        .getOrCreate()
      //spark.sparkContext.addJar("/opt/n_001_workspaces/bigdata/spark-scala-maven-2.4.0/target/spark-scala-maven-2.4.0-1.0-SNAPSHOT.jar")
      //import spark.implicits._
      spark
    }else{
      val spark = SparkSession.builder
        .master(master)
        .appName(appName)
        .config("spark.eventLog.enabled","true")
        .config("spark.history.fs.logDirectory","hdfs://standalone.com:9000/spark/log/historyEventLog")
        .config("spark.eventLog.dir","hdfs://standalone.com:9000/spark/log/historyEventLog")
        .getOrCreate()
     // spark.sparkContext.addJar("/opt/n_001_workspaces/bigdata/spark-scala-maven-2.4.0/target/spark-scala-maven-2.4.0-1.0-SNAPSHOT.jar")
      //import spark.implicits._
      spark
    }

  }

textFile

  • 读取本地文件
val spark = sparkSession(true)
    //返回dataFrame
    val df = spark.read.textFile("file:///"+ getProjectPath +"/src/main/resource/data/text/people.txt")
    df.show()

//    +-----------+
//    |      value|
//    +-----------+
//    |Michael, 29|
//    |   Andy, 30|
//    | Justin, 19|
//    |  Think, 30|
//    +-----------+

textFile

  • 读取HDFS文件
val spark = sparkSession(true)
    //返回dataFrame
    val df = spark.read.textFile("hdfs://standalone.com:9000/home/liuwen/data/people.txt")
    df.show()


//    +-----------+
//    |      value|
//    +-----------+
//    |Michael, 29|
//    |   Andy, 30|
//    | Justin, 19|
//    |  Think, 30|
//    +-----------+

    spark.stop()

text

  • 读取本地文件
val spark = sparkSession(true)
    //返回dataFrame
    val df = spark.read.text("file:///"+ getProjectPath +"/src/main/resource/data/text/people.txt")
    df.show()

//    +-----------+
//    |      value|
//    +-----------+
//    |Michael, 29|
//    |   Andy, 30|
//    | Justin, 19|
//    |  Think, 30|
//    +-----------+

text

  • 读取HDFS数据
object Run extends BaseSparkSession{

  def main(args: Array[String]): Unit = {

    val spark = sparkSession(true)
    //返回dataFrame
    val df = spark.read.text("hdfs://standalone.com:9000/home/liuwen/data/people.txt")
    df.show()

//    +-----------+
//    |      value|
//    +-----------+
//    |Michael, 29|
//    |   Andy, 30|
//    | Justin, 19|
//    |  Think, 30|
//    +-----------+

    spark.stop()
  }

}

foreach 遍历文件内容

  • 对象遍历
object Run1 extends BaseSparkSession{

  case class Person(name: String, age: Long)


  def main(args: Array[String]): Unit = {

    val spark = sparkSession(true)

    import spark.implicits._
    spark.read.textFile("file:///"+ getProjectPath +"/src/main/resource/data/text/people.txt")
      .map(line => Person(line.split(",")(0),line.split(" ")(1).trim.toLong))
        .foreach( person => println(s"name:${person.name}\t age:${person.age}"))

    spark.stop()

  }
}

first

  • 获得dataSet的第一个元素
val spark = sparkSession()
    val dataSet = spark.read.textFile("/home/liuwen/data/a.txt")

    println(dataSet.first()) //first里边调用的是head()
    spark.stop()

head

  • 获得dataSet的第一个元素
val spark = sparkSession()
    val dataSet = spark.read.textFile("/home/liuwen/data/a.text")
    println(dataSet.head()) //first里边调用的是head()

head n

  • 获得dataSet的前n个元素
val spark = sparkSession()
    val dataSet = spark.read.textFile("/home/liuwen/data/a.text")
    println(dataSet.head(5)) //first里边调用的是head()

count

  • 获得dataSet 一共有多少行数据
val spark = sparkSession()
    val dataSet = spark.read.textFile("/home/liuwen/data/a.text")
    println(dataSet.count())

collect

  • 收集dataSet中全部行的数据，在本地输出
val spark = sparkSession()
    val dataSet = spark.read.textFile("/home/liuwen/data/a.txt")
    println(dataSet.collect().mkString("\n"))

collectAsList

  • 收集dataSet中全部的数据，转成java.util.List对象
val spark = sparkSession(true)

    val dataSet = spark.read.textFile("/home/liuwen/data/a.txt")
    println( dataSet.collectAsList())
    import scala.collection.JavaConversions._
    for( v <- dataSet.collectAsList()) println(v)
    spark.stop()

foreache

  • 遍历dataSet中的每一行数据
val spark = sparkSession(true)
    val dataSet = spark.read.textFile("/home/liuwen/data/a.txt")
    dataSet.foreach(println(_))

foreache class

  • 以对象形式遍历dataSet中全部的数据
object Run1 extends BaseSparkSession{

  case class Person(name: String, age: Long)


  def main(args: Array[String]): Unit = {

    val spark = sparkSession(true)

    import spark.implicits._
    spark.read.textFile("file:///"+ getProjectPath +"/src/main/resource/data/text/people.txt")
      .map(line => Person(line.split(",")(0),line.split(" ")(1).trim.toLong))
        .foreach( person => println(s"name:${person.name}\t age:${person.age}"))

    spark.stop()


  }
}

map

  • 遍历数据集中的每个元素，进行map函数操做
val spark = sparkSession()

    val dataSet = spark.read.textFile("/home/liuwen/data/a.text")
    import spark.implicits._
    val lineWordLength = dataSet.map( line => line.split(" ").size)

    println(lineWordLength.collect().mkString("\n"))

reduce

  • 遍历dataSet中的元素，每两两进行reduce函数操做
val spark = sparkSession()

    val dataSet = spark.read.textFile("/home/liuwen/data/word.text")

    /**
      * 统计全部行单词个数
      */
    import spark.implicits._
    val lineWordLength = dataSet.map( line => line.split(" ").size)
    val result = lineWordLength.reduce((a,b) => a + b)

    println(result)

show

  • 以表格形式显示dataSet数据，默认显示前20行数据
val spark = sparkSession()

    val dataSet = spark.read.textFile("/home/liuwen/data/word.big.cn.text")

 
    val result = dataSet.show()
    println(result)

show n

  • 以表格形式显示dataSet数据，默认显示前20行数据
val spark = sparkSession()

    val dataSet = spark.read.textFile("/home/liuwen/data/word.big.cn.text")

    /**
      * 以表格的形式显示前3行数据
      * numRows是显示前几行的数据
      */

    val result = dataSet.show(3)
    println(result)

show truncate

  • 以表格形式显示dataSet数据，默认显示前20行数据
  • 参数truncate=false，是不截断显示全部数据，true是进截断
val spark = sparkSession()

    val dataSet = spark.read.textFile("/home/liuwen/data/word.big.text")

    /**
      * 以表格的形式显示前3行数据
      * numRows是显示前几行的数据
      * false 不进行返回行数据截断
      */

    val result = dataSet.show(10,false)
    println(result)

take

  • take 是至关于head
val spark = sparkSession()

    val dataSet = spark.read.textFile("/home/liuwen/data/word.big.txt")
    val result = dataSet.take(10) //等于head(n)
    println(result.mkString("\n"))

describe

val spark = sparkSession()

    val dataSet = spark.read.json("hdfs://standalone.com:9000/home/liuwen/data/json/people.json")

    dataSet.describe("name","age").show()

//    +-------+-------+------------------+
//    |summary|   name|               age|
//    +-------+-------+------------------+
//    |  count|      3|                 2|
//    |   mean|   null|              24.5|
//    | stddev|   null|7.7781745930520225|
//    |    min|   Andy|                19|
//    |    max|Michael|                30|
//    +-------+-------+------------------+

endjava

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程