大数据导论思维第12章-SPARK-SQL课件

大数据导论第十二章大数据导论第十二章1CONTENTS目录PART01SPARKSQL简介PART02SPARKSQL执行流程PART03基础数据模型DATAFRAMEPART04使用SparkSQL的方式PART05SPARKSQL数据源PART06SPARKSQLCLI介绍PART07在Pyspark中使用SparkSQLPART08在Java中连接SparkSQLPART09习题CONTENTS目录PART01SPARKSQL简介2PART01SparkSQL简介SparkSQL是一个用来处理结构化数据的Spark组件，为Spark提供了查询结构化数据的能力。SparkSQL可被视为一个分布式的SQL查询引擎，可以实现对多种数据格式和数据源进行SQL操作，包括Parquet，Hive，MongoDB，JSON、HDFS、JDBC、S3和RDD等。PART01SparkSQL简介SparkSQL3SparkSQL简介SparkSQL介绍：1.SparkSQL是为了处理结构化数据的一个Spark模块。2.不同于SparkRDD的基本API，SparkSQL接口拥有更多关于数据结构本身与执行计划等更多信息。3.在Spark内部，SparkSQL可以利用这些信息更好地对操作进行优化。4.SparkSQL提供了三种访问接口：SQL，DataFrameAPI和DatasetAPI。5.当计算引擎被用来执行一个计算时，有不同的API和语言种类可供选择。6.这种统一性意味着开发人员可以来回轻松切换各种最熟悉的API来完成同一个计算工作。SparkSQL简介SparkSQL介绍：4SparkSQL简介SparkSQL具有如下特点p数据兼容方面：能加载和查询来自各种来源的数据。p性能优化方面：除了采取内存列存储、代码生成等优化技术外，还引进成本模型对查询进行动态评估、获取最佳物理计划等；p组件扩展方面：无论是SQL的语法解析器、分析器还是优化器都可以重新定义，进行扩展。p标准连接：SparkSQL包括具有行业标准JDBC和ODBC连接的服务器模式。SparkSQL简介SparkSQL具有如下特点5SparkSQL简介SparkSQL具有如下特点p集成：无缝地将SQL查询与Spark程序混合。SparkSQL允许将结构化数据作为Spark中的分布式数据集（RDD）进行查询，在Python，Scala和Java中集成了API。这种紧密的集成使得SQL查询以及复杂的分析算法可以轻松地运行。p可扩展性：对于交互式查询和长查询使用相同的引擎。SparkSQL利用RDD模型来支持查询容错，使其能够扩展到大型作业，不需担心为历史数据使用不同的引擎。SparkSQL简介SparkSQL具有如下特点6PART02SparkSQL执行流程PART02SparkSQL执行流程7SparkSQL执行流程类似于关系型数据库，SparkSQL语句也是由Projection（a1，a2，a3）、DataSource（tableA）、Filter（condition）三部分组成，分别对应SQL查询过程中的Result、DataSource、Operation，也就是说SQL语句按Result-DataSource-Operation的次序来描述的。SparkSQL执行流程类似于关系型数据库，SparkS8SparkSQL执行流程1.解析（Parse）对读入的SQL语句进行解析，分辨出SQL语句中哪些词是关键词（如SELECT、FROM、WHERE），哪些是表达式、哪些是Projection、哪些是DataSource等，从而判断SQL语句是否规范；2.绑定（Bind）将SQL语句和数据库的数据字典（列、表、视图等）进行绑定，如果相关的Projection、DataSource等都存在，则这个SQL语句是可以执行的；SparkSQL执行流程解析（Parse）9SparkSQL执行流程3.优化（Optimize）一般的数据库会提供几个执行计划，这些计划一般都有运行统计数据，数据库会在这些计划中选择一个最优计划；4.执行（Execute）按Operation-DataSource-Result的次序来执行计划。在执行过程有时候甚至不需要读取物理表就可以返回结果，比如重新运行刚运行过的SQL语句，可能直接从数据库的缓冲池中获取返回结果。SparkSQL执行流程优化（Optimize）10PART03基础数据模型DataFrameDataFrame是由“命名列”（类似关系表的字段定义）所组织起来的一个分布式数据集合，可以把它看成是一个关系型数据库的表。PART03基础数据模型DataFrameDataFra11基础数据模型DataFrameDataFrame是SparkSQL的核心，它将数据保存为行构成的集合，行对应列有相应的列名。DataFrame与RDD的主要区别在于，DataFrame带有Schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。这使得SparkSQL可以掌握更多的结构信息，从而能够对DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化，最终达到大幅提升运行时效率的目标。基础数据模型DataFrameDataFrame 是Spark12基础数据模型DataFrameDataFrame与RDD的对比：基础数据模型DataFrameDataFrame 与RDD的对13PART04使用SparkSQL的方式PART04使用SparkSQL的方式14使用SparkSQL的方式使用SparkSQL，首先利用sqlContext从外部数据源加载数据为DataFrame；然后，利用DataFrame上丰富的API进行查询、转换；最后，将结果进行展现或存储为各种外部数据形式。SparkSQL为Spark提供了查询结构化数据的能力，查询时既可以使用SQL也可以使用DataFrameAPI（RDD）。通过ThriftServer，SparkSQL支持多语言编程包括Java、Scala、Python及R。使用SparkSQL的方式使用SparkSQL，首先利用15使用SparkSQL的方式使用SparkSQL的方式16使用SparkSQL的方式1.加载数据.从Hive中的users表构造DataFrame：users=sqlContext.table(users).加载S3上的JSON文件：logs=sqlContext.load(s3n:/path/to/data.json,json).加载HDFS上的Parquet文件：clicks=sqlContext.load(hdfs:/path/to/data.parquet,parquet)使用SparkSQL的方式加载数据.从Hive中的us17使用SparkSQL的方式1.加载数据.通过JDBC访问MySQL：comments=sqlContext.jdbc(jdbc:mysql:/localhost/comments,user).将普通RDD转变为DataFrame：rdd=sparkContext.textFile(“article.txt”).flatMap(_.split().map(_,1).reduceByKey(_+_)wordCounts=sqlContext.createDataFrame(rdd,word,count)使用SparkSQL的方式加载数据.通过JDBC访问M18使用SparkSQL的方式1.加载数据.将本地数据容器转变为DataFrame：data=(Alice,21),(Bob,24)people=sqlContext.createDataFrame(data,name,age).将PandasDataFrame转变为SparkDataFrame（PythonAPI特有功能）：sparkDF=sqlContext.createDataFrame(pandasDF）使用SparkSQL的方式加载数据.将本地数据容器转变19使用SparkSQL的方式2.使用DataFrame.创建一个只包含年轻用户的DataFrame：young=users.filter(users.age21).也可以使用Pandas风格的语法：young=usersusers.age=13ANDage=19)teenagers.show()Parquet文件数据源读取上面存储的Parquet文件为DataFrameParq32JSONDataSets数据源JSONDataSets数据源SparkSQL可以自动根据JSONDataSet的格式把其上载为DataFrame。用路径指定JSONdataset；路径下可以是一个文件，也可以是多个文件：sc=spark.sparkContextpath=examples/src/main/resources/people.jsonpeopleDF=spark.read.json(path)使用的结构可以调用printSchema()方法打印：peopleDF.printSchema()JSONDataSets数据源JSONDataSets33利用DataFrame创建一个临时表：使用Spark的sql方法进行SQL查询：peopleDF.createOrReplaceTempView(people)teenagerNamesDF=spark.sql(SELECTnameFROMpeopleWHEREageBETWEEN13AND19)teenagerNamesDF.show()JSONDataSets数据源利用DataFrame创建一个临时表：使用Spark的sql34JSONdataset的DataFrame也可以是RDDString格式，每个JSON对象为一个string：jsonStrings=name:Yin,address:city:Columbus,state:OhiootherPeopleRDD=sc.parallelize(jsonStrings)otherPeople=spark.read.json(otherPeopleRDD)otherPeople.show()JSONDataSets数据源JSONdataset的DataFrame也可以是RDD35Hive表数据源Hive表数据源SparkSQL支持对Hive中的数据进行读写。首先创建一个支持Hive的SparkSession对象，包括与Hivemetastore的连接，支持Hive的序列化和反序列化操作，支持用户定义的Hive操作等。warehouse_location=abspath(spark-warehouse)spark=SparkSession.builder.appName(PythonSparkSQLHiveintegrationexample).config(spark.sql.warehouse.dir,warehouse_location).enableHiveSupport().getOrCreate()warehouse_location指定数据库和表的缺省位置：Hive表数据源Hive表数据源SparkSQL支持对Hi36Hive表数据源spark.sql(CREATETABLEIFNOTEXISTSsrc(keyINT,valueSTRING)USINGhive)spark.sql(LOADDATALOCALINPATHexamples/src/main/resources/kv1.txtINTOTABLEsrc)基于新创建的SparkSession创建表和上载数据到表中：spark.sql(SELECT*FROMsrc).show()spark.sql(SELECTCOUNT(*)FROMsrc).show()使用HiveQL进行查询：Hive表数据源spark.sql(CREATETABL37Hive表数据源sqlDF=spark.sql(SELECTkey,valueFROMsrcWHEREkeyvalsqlContext=neworg.apache.spark.sql.SQLContext(sc)sqlContext:org.apache.spark.sql.SQLContext=org.apache.spark.sql.SQLContext1943a343scalaimportsqlContext.implicits._importsqlContext.implicits._1.SQLContextSparkSQLCLI介绍SparkSQL所有功能的入43SparkSQLCLI介绍下面的操作基于一个简单的数据文件people.json，文件的内容如下：name:Michaelname:Andy,age:30name:Justin,age:192.数据文件下面语句从本地文件people.json读取数据创建DataFrame：valdf=sqlContext.read.json(file:/data/people.json)df:org.apache.spark.sql.DataFrame=age:bigint,name:string3.创建DataFramesSparkSQLCLI介绍下面的操作基于一个简单的数据文44Pyspark是针对Spark的PythonAPI。Spark使用py4j来实现Python与Java的互操作，从而实现使用Python编写Spark程序。Spark也同样提供了Pyspark，一个Spark的PythonShell，可以以交互的方式使用Python编写Spark程序。PART07在Pyspark中使用SparkSQLPyspark是针对Spark的PythonAPI。Spa45在Pyspark中使用SparkSQL在终端上启动PythonSparkShell：./bin/pyspark使用JSON文件作为数据源，创建JSON文件/home/sparksql/courses.json，并输入下面的内容：1.实例描述name:Linux,type:basic,length:10name:TCPIP,type:project,length:15name:Python,type:project,length:8name:GO,type:basic,length:2name:Ruby,type:basic,length:5在Pyspark中使用SparkSQL在终端上启动Pyth46在Pyspark中使用SparkSQL首先使用SQLContext模块，其作用是提供SparkSQL处理的功能。在PysparkShell中逐步输入下面步骤的内容：引入pyspark.sql中的SQLContext：frompyspark.sqlimportSQLContext2.创建SQLContext对象使用pyspark的SparkContext对象，创建SQLContext对象：sqlContext=SQLContext(sc)在Pyspark中使用SparkSQL首先使用SQLCon47在Pyspark中使用SparkSQLDataFrame对象可以由RDD创建，也可以从Hive表或JSON文件等数据源创建。创建DataFrame，指明来源自JSON文件：df=sqlContext.read.json(/home/shiyanlou/courses.json)3.创建DataFrame对象在Pyspark中使用SparkSQLDataFrame对48在Pyspark中使用SparkSQL首先打印当前DataFrame里的内容和数据表的格式：df.select(name).show()#展示了所有的课程名df.select(name,length).show()#展示了所有的课程名及课程长度4.对DataFrame进行操作show()函数将打印出JSON文件中存储的数据表；使用printSchema()函数打印数据表的格式。然后对DataFrame的数据进行各种操作：df.show()df.printSchema()在Pyspark中使用SparkSQL首先打印当前Data49在Pyspark中使用SparkSQLdf.filter(dftype=basic).select(name,type).show()#展示了课程类型为基础课（basic）的课程名和课程类型df.groupBy(type).count().show()#计算所有基础课和项目课的数量。首先需要将DataFrame注册为Table才可以在该表上执行SQL语句：df.registerTempTable(courses)coursesRDD=sqlContext.sql(SELECTnameFROMcoursesWHERElength=5andlength=10)names=coursesRDD.rdd.map(lambdap:Name:+p.name)fornameinnames.collect():printname5.执行SQL语句在Pyspark中使用SparkSQLdf.filter(50在Pyspark中使用SparkSQLParquet是SparkSQL读取的默认数据文件格式，把从JSON中读取的DataFrame保存为Parquet格式，只保存课程名称和长度两项数据：df.select(name,length).write.save(/tmp/courses.parquet,format=parquet)6.保存DataFrame为其他格式将创建hdfs:/master:9000/tmp/courses.parquet文件夹并存入课程名称和长度数据。在Pyspark中使用SparkSQLParquet是Sp51SparkSQL实现了ThriftJDBC/ODBCserver，所以Java程序可以通过JDBC远程连接SparkSQL发送SQL语句并执行。PART08在Java中连接SparkSQLSparkSQL实现了ThriftJDBC/ODBCs52在Java中连接SparkSQL首先将$HIVE_HOME/conf/hive-site.xml拷贝到$SPARK_HOME/conf目录下。另外，因为Hive元数据信息存储在MySQL中，所以Spark在访问这些元数据信息时需要MySQL连接驱动的支持。添加驱动的方式有三种：在$SPARK_HOME/conf目录下的spark-defaults.conf中添加：spark.jars/opt/lib2/mysql-connector-java-5.1.26-bin.jar；可以实现添加多个依赖jar比较方便：spark.driver.extraClassPath/opt/lib2/mysql-connector-java-5.1.26-bin.jar；1.设置配置在Java中连接SparkSQL首先将$HIVE_HOM53在运行时添加-jars/opt/lib2/mysql-connector-java-5.1.26-bin.jar做完上面的准备工作后，SparkSQL和Hive就继承在一起了，SparkSQL可以读取Hive中的数据。1.设置配置2.启动Thrift在Spark根目录下执行：./sbin/start-thriftserver.sh开启thrift服务器，它可以接受所有spark-submit的参数，并且还可以接受-hiveconf参数。不添加任何参数表示以local方式运行，默认的监听端口为10000在Java中连接SparkSQL在运行时添加-jars/opt/lib2/mysql-543.添加依赖打开Eclipse用JDBC连接HiveServer2。新建一个Maven项目，在pom.xml添加以下依赖：org.apache.hivehive-jdbc1.2.1org.apache.hadoophadoop-common2.4.1在Java中连接SparkSQL添加依赖打开Eclipse用JDBC连接HiveServe553.添加依赖jdk.toolsjdk.tools1.6system$JAVA_HOME/lib/tools.jar在Java中连接SparkSQL添加依赖在Java中连接Spar56JDBC连接HiveServer2的相关参数：l驱动：org.apache.hive.jdbc.HiveDriverlurl：jdbc:hive2:/192.168.1.131:10000/defaultl用户名：hadoop(启动thriftserver的linux用户名)l密码：“”（默认密码为空）4.JDBC连接参数在Java中连接SparkSQLJDBC连接HiveServer2的相关参数：JDBC连接57importjava.sql.Connection;importjava.sql.DriverManager;importjava.sql.ResultSet;importjava.sql.SQLException;importjava.sql.Statement;publicclassTest1publicstaticvoidmain(Stringargs)throwsSQLExceptionStringurl=jdbc:hive2:/192.168.1.131:10000/default;tryClass.forName(org.apache.hive.jdbc.HiveDriver);catch(ClassNotFoundExceptione)e.printStackTrace();Connectionconn=DriverManager.getConnection(url,hadoop,);Statementstmt=conn.createStatement();Stringsql=SELECT*FROMdoc1limit10;System.out.println(Running+sql);ResultSetres=stmt.executeQuery(sql);while(res.next()System.out.println(id:+res.getInt(1)+ttype:+res.getString(2)+tauthors:+res.getString(3)+ttitle:+res.getString(4)+tyear:+res.getInt(5);5.JDBC连接参数在Java中连接SparkSQLimportjava.sql.Connection;J58PART09作业PART09作业59作业作业：1.什么是SparkSQL？其主要目的是什么？2.SparkSQL的执行流程有哪几个步骤？3.在SparkSQL中，什么是DataFrame？使用DataFrame的优势是什么？DataFrame与RDD的主要区别是什么？4.使用SparkSQL的方式有哪几种？使用SparkSQL的步骤是什么？5.常用的SparkSQL的数据源有哪些？6.Parquet文件格式是什么？它的主要特点是什么？7.为了使Java程序可以通过JDBC远程连接SparkSQL，需要做哪些准备工作？连接数据库的语句是什么？有哪些参数？作业作业：什么是SparkSQL？其主要目的是什么？60作业作业：8.请按下述要求写出相应的SparkSQL语句：从一个本地JSON文件创建DataFrame；打印DataFrame元数据；按照列属性过滤DataFrame的数据；返回某列满足条件的数据；把DataFrame注册成数据库表。作业作业：请按下述要求写出相应的SparkSQL语句：61谢谢FOR YOUR LISTENINGHandge CO.LTD.2016.12.09谢谢FORYOURLISTENINGHandgeCO.62