How to use SparkSession in Apache Spark 2.0

相似文件 换一批

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

1. Generally, a session is an interaction between two or more entities. In computer parlance, its usage is prominent in the realm of networked computers on the internet. First with TCP session, then with login session, followed by HTTP and user session, so no surprise that we now have SparkSession, introduced in Apache Spark 2.0.
   
2. 
   
3. Beyond a time-bounded interaction, SparkSession provides a single point of entry to interact with underlying Spark functionality and allows programming Spark with DataFrame and Dataset APIs. Most importantly, it curbs the number of concepts and constructs a developer has to juggle while interacting with Spark.
   
4. 
   
5. In this blog and its accompanying Databricks notebook, we will explore SparkSession functionality in Spark 2.0.

复制代码

本帖隐藏的内容

How to use SparkSession in Apache Spark 2.0.pdf (428.77 KB)

2017-5-28 02:00:20 上传

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏1 回帖

关键词：Apache Spark session apache Sparks Spark

1. Creating a SparkSession
   
2. In previous versions of Spark, you had to create a SparkConf and SparkContext to interact with Spark, as shown here:
   
3. 
   
4. //set up the spark configuration and create contexts
   
5. val sparkConf = new SparkConf().setAppName("SparkSessionZipsExample").setMaster("local")
   
6. // your handle to SparkContext to access other context like SQLContext
   
7. val sc = new SparkContext(sparkConf).set("spark.some.config.option", "some-value")
   
8. val sqlContext = new org.apache.spark.sql.SQLContext(sc)

复制代码

1. Whereas in Spark 2.0 the same effects can be achieved through SparkSession, without expliciting creating SparkConf, SparkContext or SQLContext, as they’re encapsulated within the SparkSession. Using a builder design pattern, it instantiates a SparkSession object if one does not already exist, along with its associated underlying contexts.
   
2. 
   
3. // Create a SparkSession. No need to create SparkContext
   
4. // You automatically get it as part of the SparkSession
   
5. val warehouseLocation = "file:${system:user.dir}/spark-warehouse"
   
6. val spark = SparkSession
   
7.    .builder()
   
8.    .appName("SparkSessionZipsExample")
   
9.    .config("spark.sql.warehouse.dir", warehouseLocation)
   
10.    .enableHiveSupport()
    
11.    .getOrCreate()
    
12. At this point you can use the spark variable as your instance object to access its public methods and instances for the duration of your Spark job.

复制代码

1. Configuring Spark’s Runtime Properties
   
2. Once the SparkSession is instantiated, you can configure Spark’s runtime config properties. For example, in this code snippet, we can alter the existing runtime config options. Since configMap is a collection, you can use all of Scala’s iterable methods to access the data.
   
3. 
   
4. //set new runtime options
   
5. spark.conf.set("spark.sql.shuffle.partitions", 6)
   
6. spark.conf.set("spark.executor.memory", "2g")
   
7. //get all settings
   
8. val configMap:Map[String, String] = spark.conf.getAll()

复制代码

1. Accessing Catalog Metadata
   
2. Often, you may want to access and peruse the underlying catalog metadata. SparkSession exposes “catalog” as a public instance that contains methods that work with the metastore (i.e data catalog). Since these methods return a Dataset, you can use Dataset API to access or view data. In this snippet, we access table names and list of databases.
   
3. 
   
4. //fetch metadata data from the catalog
   
5. spark.catalog.listDatabases.show(false)
   
6. spark.catalog.listTables.show(false)

复制代码

1. Creating Datasets and Dataframes
   
2. There are a number of ways to create DataFrames and Datasets using SparkSession APIs
   
3. One quick way to generate a Dataset is by using the spark.range method. When learning to manipulate Dataset with its API, this quick method proves useful. For example,
   
4. 
   
5. //create a Dataset using spark.range starting from 5 to 100, with increments of 5
   
6. val numDS = spark.range(5, 100, 5)
   
7. // reverse the order and display first 5 items
   
8. numDS.orderBy(desc("id")).show(5)
   
9. //compute descriptive stats and display them
   
10. numDs.describe().show()
    
11. // create a DataFrame using spark.createDataFrame from a List or Seq
    
12. val langPercentDF = spark.createDataFrame(List(("Scala", 35), ("Python", 30), ("R", 15), ("Java", 20)))
    
13. //rename the columns
    
14. val lpDF = langPercentDF.withColumnRenamed("_1", "language").withColumnRenamed("_2", "percent")
    
15. //order the DataFrame in descending order of percentage
    
16. lpDF.orderBy(desc("percent")).show(false)

复制代码

1. Reading JSON Data with SparkSession API
   
2. Like any Scala object you can use spark, the SparkSession object, to access its public methods and instance fields. I can read JSON or CVS or TXT file, or I can read a parquet table. For example, in this code snippet, we will read a JSON file of zip codes, which returns a DataFrame, a collection of generic Rows.
   
3. 
   
4. // read the json file and create the dataframe
   
5. val jsonFile = args(0)
   
6. val zipsDF = spark.read.json(jsonFile)
   
7. //filter all cities whose population > 40K
   
8. zipsDF.filter(zipsDF.col("pop") > 40000).show(10)

复制代码

1. Using Spark SQL with SparkSession
   
2. Through SparkSession, you can access all of the Spark SQL functionality as you would through SQLContext. In the code sample below, we create a table against which we issue SQL queries.
   
3. 
   
4. // Now create an SQL table and issue SQL queries against it without
   
5. // using the sqlContext but through the SparkSession object.
   
6. // Creates a temporary view of the DataFrame
   
7. zipsDF.createOrReplaceTempView("zips_table")
   
8. zipsDF.cache()
   
9. val resultsDF = spark.sql("SELECT city, pop, state, zip FROM zips_table")
   
10. resultsDF.show(10)

复制代码

1. Saving and Reading from Hive table with SparkSession
   
2. Next, we are going to create a Hive table and issue queries against it using SparkSession object as you would with a HiveContext.
   
3. 
   
4. //drop the table if exists to get around existing table error
   
5. spark.sql("DROP TABLE IF EXISTS zips_hive_table")
   
6. //save as a hive table
   
7. spark.table("zips_table").write.saveAsTable("zips_hive_table")
   
8. //make a similar query against the hive table
   
9. val resultsHiveDF = spark.sql("SELECT city, pop, state, zip FROM zips_hive_table WHERE pop > 40000")
   
10. resultsHiveDF.show(10)

复制代码

我看看谢谢啊