spark2的SparkSession思考与总结2：SparkSession包含哪些函数及功能介绍

问题导读

1.spark SparkSession包含哪些函数？
2.创建DataFrame有哪些函数？
3.创建DataSet有哪些函数?




上一篇spark2：SparkSession思考与总结1
http://www.aboutyun.com/forum.php?mod=viewthread&tid=23381


版本：spark2我们在学习的过程中，很多都是注重实战，这没有错的，但是如果在刚开始入门就能够了解这些函数，在遇到新的问题，可以找到方向去解决问题。比如我们常用的创建DateFrame和DataTable方式就那么一种或则两种，如果更多那就看不懂了。在比如想测试下程序的性能，这时候如果自己写，那就太麻烦了，可以使用spark提供的Time函数。这就是知识全面的一个好处。

SparkSession是一个比较重要的类，它的功能的实现，肯定包含比较多的函数，这里介绍下它包含哪些函数。
builder函数
public static SparkSession.Builder builder()
创建 SparkSession.Builder，初始化SparkSession.

setActiveSession函数
public static void setActiveSession(SparkSession session)
当SparkSession.GetOrCreate()被调用，SparkSession发生变化，将会返回一个线程和它的子线程。这将会确定给定的线程接受带有隔离会话的SparkSession，而不是全局的context。

clearActiveSession函数
public static void clearActiveSession()
清除当前线程的Active SparkSession。然后调用GetOrCreate将会返回第一次创建的context代替本地线程重写

setDefaultSession函数
public static void setDefaultSession(SparkSession session)
设置默认的SparkSession，返回builder

clearDefaultSession函数
public static void clearDefaultSession()
清除默认的SparkSession返回的builder

getActiveSession函数
public static scala.Option<SparkSession> getActiveSession()
由builder，返回当前线程的Active SparkSession

getDefaultSession函数

public static scala.Option<SparkSession> getDefaultSession()
由builder，返回默认的SparkSession


sparkContext函数
public SparkContext sparkContext()


version函数
public String version()
返回运行应用程序的spark版本

sharedState函数
public org.apache.spark.sql.internal.SharedState sharedState()
通过sessions共享状态，包括SparkContext, cached 数据, listener, 和catalog.
这是内部spark，接口稳定性没有保证

sessionState函数
public org.apache.spark.sql.internal.SessionState sessionState()
通过session隔离状态，包括：SQL 配置, 临时表, registered 功能, 和 其它可接受的 SQLConf.
这是内部spark，接口稳定性没有保证


sqlContext函数
public SQLContext sqlContext()
session封装以 SQLContext的形式，为了向后兼容。

conf函数
public RuntimeConfig conf()
运行spark 配置接口
通过这个接口用户可以设置和获取与spark sql相关的所有Spark 和Hadoop配置.当获取config值，

listenerManager函数
public ExecutionListenerManager listenerManager()
用于注册自定义QueryExecutionListeners的接口，用于侦听执行指标。

experimental函数
public ExperimentalMethods experimental()
collection函数，被认为是experimental，可以用于查询高级功能的查询计划程序。

udf函数
public UDFRegistration udf()
collection 函数，用于用户自定义函数
例子：
Scala版本：
[mw_shl_code=scala,true]   sparkSession.udf.register("myUDF", (arg1: Int, arg2: String) => arg2 + arg1)
[/mw_shl_code]

Java版本：
[mw_shl_code=java,true] sparkSession.udf().register("myUDF",
       (Integer arg1, String arg2) -> arg2 + arg1,
       DataTypes.StringType);[/mw_shl_code]

streams函数
public StreamingQueryManager streams()
返回StreamingQueryManager ，允许管理所有的StreamingQuerys

newSession函数
public SparkSession newSession()
启动一个独立的 SQL 配置, temporary 表, registered 功能新的session，但共享底层的SparkContext 和缓存数据.

emptyDataFrame函数
public Dataset<Row> emptyDataFrame()
返回一个空没有行和列的DataFrame

emptyDataset函数
public <T> Dataset<T> emptyDataset(Encoder<T> evidence$1)
创建一个T类型的空的Dataset

createDataFrame函数
public <A extends scala.Product> Dataset<Row> createDataFrame(RDD<A> rdd,scala.reflect.api.TypeTags.TypeTag<A> evidence$2)
从rdd创建DateFrame

public Dataset<Row> createDataFrame(RDD<Row> rowRDD, StructType schema)
从RDD包含的行给定的schema，创建DataFrame。需要确保每行的RDD结构匹配提供的schema，否则将会运行异常。例如：
[mw_shl_code=scala,true]import org.apache.spark.sql._
  import org.apache.spark.sql.types._
  val sparkSession = new org.apache.spark.sql.SparkSession(sc)

  val schema =
    StructType(
      StructField("name", StringType, false) ::
      StructField("age", IntegerType, true) :: Nil)

  val people =
    sc.textFile("examples/src/main/resources/people.txt").map(
      _.split(",")).map(p => Row(p(0), p(1).trim.toInt))
  val dataFrame = sparkSession.createDataFrame(people, schema)
  dataFrame.printSchema
  // root
  // |-- name: string (nullable = false)
  // |-- age: integer (nullable = true)

  dataFrame.createOrReplaceTempView("people")
  sparkSession.sql("select name from people").collect.foreach(println)[/mw_shl_code]

public Dataset<Row> createDataFrame(JavaRDD<Row> rowRDD,StructType schema)
创建DataFrame从包含schema的行的RDD。确保RDD提供的每行结构匹配提供的schema，否则运行异常
public Dataset<Row> createDataFrame(java.util.List<Row> rows,StructType schema)

创建DataFrame从包含行的schema的java.util.List

public Dataset<Row> createDataFrame(RDD<?> rdd,Class<?> beanClass)
应用schema到Java Beans的RDD
警告：由于Java Bean中的字段没有保证的顺序，因此SELECT *查询将以未定义的顺序返回列。

public Dataset<Row> createDataFrame(JavaRDD<?> rdd, Class<?> beanClass)
应用schema到Java Beans的RDD
警告：由于Java Bean中的字段没有保证的顺序，因此SELECT *查询将以未定义的顺序返回列。
public Dataset<Row> createDataFrame(java.util.List<?> data,Class<?> beanClass)

应用schema到Java Bean list
警告：由于Java Bean中的字段没有保证的顺序，因此SELECT *查询将以未定义的顺序返回列。

baseRelationToDataFrame函数
public Dataset<Row> baseRelationToDataFrame(BaseRelation baseRelation)
转换创建的BaseRelation，为外部数据源到DataFrame


createDataset函数
public <T> Dataset<T> createDataset(scala.collection.Seq<T> data,Encoder<T> evidence$4)
从本地给定类型的数据Seq创建DataSet。这个方法需要encoder （将T类型的JVM对象转换为内部Spark SQL表示形式）。这通常是通过从sparksession implicits自动创建。或则可以通过调用 Encoders上的静态方法来显式创建。
例子：
[mw_shl_code=scala,true] import spark.implicits._
   case class Person(name: String, age: Long)
   val data = Seq(Person("Michael", 29), Person("Andy", 30), Person("Justin", 19))
   val ds = spark.createDataset(data)

   ds.show()
   // +-------+---+
   // |   name|age|
   // +-------+---+
   // |Michael| 29|
   // |   Andy| 30|
   // | Justin| 19|
   // +-------+---+[/mw_shl_code]


public <T> Dataset<T> createDataset(RDD<T> data,Encoder<T> evidence$5)
创建DataSet从给定类型的RDD。这个方法需要encoder （将T类型的JVM对象转换为内部Spark SQL表示形式）。通常自动创建通过SparkSession的implicits 或则可以通过调用 Encoders上的静态方法来显式创建。

public <T> Dataset<T> createDataset(java.util.List<T> data,Encoder<T> evidence$6)
创建 Dataset，对于T类型的java.util.List。这个方法需要encoder （将T类型的JVM对象转换为内部Spark SQL表示形式）， 或则可以通过调用 Encoders上的静态方法来显式创建。
Java例子
[mw_shl_code=java,true] List<String> data = Arrays.asList("hello", "world");
     Dataset<String> ds = spark.createDataset(data, Encoders.STRING());[/mw_shl_code]

range函数
public Dataset<Long> range(long end)使用名为id的单个LongType列创建一个Dataset，包含元素的范围从0到结束（不包括），步长值为1。

public Dataset<Long> range(long start,long end)
使用名为id的单个LongType列创建一个Dataset，包含元素的范围从start到结束（不包括），步长值为1。

public Dataset<Long> range(long start, long end, long step)
使用名为id的单个LongType列创建一个Dataset，包含元素的范围从start到结束（不包括），步长值为step。


public Dataset<Long> range(long start,long end,long step,int numPartitions)
使用名为id的单个LongType列创建一个Dataset，包含元素的范围从start到结束（不包括），步长值为step，指定partition 的数目


catalog函数
public Catalog catalog()
用户可以通过它 create, drop, alter 或则query 底层数据库, 表, 函数等.

table函数
public Dataset<Row> table(String tableName)返回指定的table/view作为DataFrame
tableName是可以合格或则不合格的名称。如果在数据库中指定，它在数据库中会识别。否则它会尝试找到一个临时view ，匹配到当前数据库的table/view，全局的临时的数据库view也是有效的。


sql函数
public Dataset<Row> sql(String sqlText)
使用spark执行sql查询，作为DataFrame返回结果。用来sql parsing，可以用spark.sql.dialect来配置

read函数
public DataFrameReader read()
返回一个DataFrameReader,可以用来读取非流数据作为一个DataFrame
[mw_shl_code=scala,true]   sparkSession.read.parquet("/path/to/file.parquet")
   sparkSession.read.schema(schema).json("/path/to/file.json")[/mw_shl_code]

readStream函数
public DataStreamReader readStream()
返回一个DataFrameReader,可以用来读取流数据作为一个DataFrame
[mw_shl_code=scala,true]  sparkSession.readStream.parquet("/path/to/directory/of/parquet/files")
   sparkSession.readStream.schema(schema).json("/path/to/directory/of/json/files")[/mw_shl_code]

time函数
public <T> T time(scala.Function0<T> f)
执行一些代码块并打印输出执行该块所花费的时间。 这仅在Scala中可用，主要用于交互式测试和调试。
这个函数还是比较有用的，很多地方都能用到

implicits函数
public SparkSession.implicits$ implicits()
嵌套Scala对象访问

stop函数
public void stop()
停止SparkContext

close函数
public void close()
与stop类似

以上内容来自官网，如有不同的理解，欢迎大家讨论指正