Spark SQL常见4种数据源详解

2023-08-21 02:15:04 189

通用load/write方法

手动指定选项

SparkSQL的DataFrame接口支持多种数据源的操作。一个DataFrame可以进行RDDs方式的操作，也可以被注册为临时表。把DataFrame注册为临时表之后，就可以对该DataFrame执行SQL查询。

SparkSQL的默认数据源为Parquet格式。数据源为Parquet文件时，SparkSQL可以方便的执行所有的操作。

修改配置项spark.sql.sources.default，可修改默认数据源格式。

scala>valdf=spark.read.load("hdfs://hadoop001:9000/namesAndAges.parquet")
df:org.apache.spark.sql.DataFrame=[age:bigint,name:string]
scala>df.select("name").write.save("names.parquet")

当数据源格式不是parquet格式文件时，需要手动指定数据源的格式。数据源格式需要指定全名（例如：org.apache.spark.sql.parquet），如果数据源格式为内置格式，则只需要指定简称json,parquet,jdbc,orc,libsvm,csv,text来指定数据的格式。

可以通过SparkSession提供的read.load方法用于通用加载数据，使用write和save保存数据。

scala>valpeopleDF=spark.read.format("json").load("hdfs://hadoop001:9000/people.json")
peopleDF:org.apache.spark.sql.DataFrame=[age:bigint,name:string]
scala>peopleDF.write.format("parquet").save("hdfs://hadoop001:9000/namesAndAges.parquet")
scala>

除此之外，可以直接运行SQL在文件上:

valsqlDF=spark.sql("SELECT*FROMparquet.`hdfs://hadoop001:9000/namesAndAges.parquet`")
sqlDF.show()

文件保存选项

可以采用SaveMode执行存储操作，SaveMode定义了对数据的处理模式。需要注意的是，这些保存模式不使用任何锁定，不是原子操作。此外，当使用Overwrite方式执行时，在输出新数据之前原数据就已经被删除。SaveMode详细介绍如下表：

Scala/Java	AnyLanguage	Meaning
SaveMode.ErrorIfExists(default)	“error”(default)	如果文件存在，则报错
SaveMode.Append	“append”	追加
SaveMode.Overwrite	“overwrite”	覆写
SaveMode.Ignore	“ignore”	数据存在，则忽略

Parquet文件

Parquet读写

Parquet格式经常在Hadoop生态圈中被使用，它也支持SparkSQL的全部数据类型。SparkSQL提供了直接读取和存储Parquet格式文件的方法。

//Encodersformostcommontypesareautomaticallyprovidedbyimportingspark.implicits._
importspark.implicits._
valpeopleDF=spark.read.json("examples/src/main/resources/people.json")
//DataFramescanbesavedasParquetfiles,maintainingtheschemainformation
peopleDF.write.parquet("hdfs://hadoop001:9000/people.parquet")
//Readintheparquetfilecreatedabove
//Parquetfilesareself-describingsotheschemaispreserved
//TheresultofloadingaParquetfileisalsoaDataFrame
valparquetFileDF=spark.read.parquet("hdfs://hadoop001:9000/people.parquet")
//ParquetfilescanalsobeusedtocreateatemporaryviewandthenusedinSQLstatements
parquetFileDF.createOrReplaceTempView("parquetFile")
valnamesDF=spark.sql("SELECTnameFROMparquetFileWHEREageBETWEEN13AND19")
namesDF.map(attributes=>"Name:"+attributes(0)).show()
//+------------+
//|value|
//+------------+
//|Name:Justin|
//+------------+

解析分区信息

对表进行分区是对数据进行优化的方式之一。在分区的表内，数据通过分区列将数据存储在不同的目录下。Parquet数据源现在能够自动发现并解析分区信息。例如，对人口数据进行分区存储，分区列为gender和country，使用下面的目录结构：

path
└──to
└──table
├──gender=male
│├──...
││
│├──country=US
││└──data.parquet
│├──country=CN
││└──data.parquet
│└──...
└──gender=female
├──...
│
├──country=US
│└──data.parquet
├──country=CN
│└──data.parquet
└──...

通过传递path/to/table给SQLContext.read.parque

或SQLContext.read.load，SparkSQL将自动解析分区信息。

返回的DataFrame的Schema如下：

root
|--name:string(nullable=true)
|--age:long(nullable=true)
|--gender:string(nullable=true)
|--country:string(nullable=true)

需要注意的是，数据的分区列的数据类型是自动解析的。当前，支持数值类型和字符串类型。自动解析分区类型的参数为：

spark.sql.sources.partitionColumnTypeInference.enabled

默认值为true。

如果想关闭该功能，直接将该参数设置为disabled。此时，分区列数据格式将被默认设置为string类型，不再进行类型解析。

Schema合并

像ProtocolBuffer、Avro和Thrift那样，Parquet也支持Schemaevolution（Schema演变）。用户可以先定义一个简单的Schema，然后逐渐的向Schema中增加列描述。通过这种方式，用户可以获取多个有不同Schema但相互兼容的Parquet文件。现在Parquet数据源能自动检测这种情况，并合并这些文件的schemas。

因为Schema合并是一个高消耗的操作，在大多数情况下并不需要，所以SparkSQL从1.5.0开始默认关闭了该功能。可以通过下面两种方式开启该功能：

当数据源为Parquet文件时，将数据源选项mergeSchema设置为true。

设置全局SQL选项：

spark.sql.parquet.mergeSchema为true。

//sqlContextfromthepreviousexampleisusedinthisexample.
//ThisisusedtoimplicitlyconvertanRDDtoaDataFrame.
importspark.implicits._
//CreateasimpleDataFrame,storedintoapartitiondirectory
valdf1=sc.makeRDD(1to5).map(i=>(i,i*2)).toDF("single","double")
df1.write.parquet("hdfs://hadoop001:9000/data/test_table/key=1")
//CreateanotherDataFrameinanewpartitiondirectory,
//addinganewcolumnanddroppinganexistingcolumn
valdf2=sc.makeRDD(6to10).map(i=>(i,i*3)).toDF("single","triple")
df2.write.parquet("hdfs://hadoop001:9000/data/test_table/key=2")
//Readthepartitionedtable
valdf3=spark.read.option("mergeSchema","true").parquet("hdfs://hadoop001:9000/data/test_table")
df3.printSchema()
//Thefinalschemaconsistsofall3columnsintheParquetfilestogether
//withthepartitioningcolumnappearedinthepartitiondirectorypaths.
//root
//|--single:int(nullable=true)
//|--double:int(nullable=true)
//|--triple:int(nullable=true)
//|--key:int(nullable=true)

Hive数据源

ApacheHive是Hadoop上的SQL引擎，SparkSQL编译时可以包含Hive支持，也可以不包含。包含Hive支持的SparkSQL可以支持Hive表访问、UDF(用户自定义函数)以及Hive查询语言(HiveQL/HQL)等。需要强调的一点是，如果要在SparkSQL中包含Hive的库，并不需要事先安装Hive。一般来说，最好还是在编译SparkSQL时引入Hive支持，这样就可以使用这些特性了。如果你下载的是二进制版本的Spark，它应该已经在编译时添加了Hive支持。

若要把SparkSQL连接到一个部署好的Hive上，你必须把hive-site.xml复制到Spark的配置文件目录中($SPARK_HOME/conf)。即使没有部署好Hive，SparkSQL也可以运行。

需要注意的是，如果你没有部署好Hive，SparkSQL会在当前的工作目录中创建出自己的Hive元数据仓库，叫作metastore_db。此外，如果你尝试使用HiveQL中的CREATETABLE(并非CREATEEXTERNALTABLE)语句来创建表，这些表会被放在你默认的文件系统中的/user/hive/warehouse目录中(如果你的classpath中有配好的hdfs-site.xml，默认的文件系统就是HDFS，否则就是本地文件系统)。

importjava.io.File
importorg.apache.spark.sql.Row
importorg.apache.spark.sql.SparkSession
caseclassRecord(key:Int,value:String)
//warehouseLocationpointstothedefaultlocationformanageddatabasesandtables
valwarehouseLocation=newFile("spark-warehouse").getAbsolutePath
valspark=SparkSession
.builder()
.appName("SparkHiveExample")
.config("spark.sql.warehouse.dir",warehouseLocation)
.enableHiveSupport()
.getOrCreate()
importspark.implicits._
importspark.sql
sql("CREATETABLEIFNOTEXISTSsrc(keyINT,valueSTRING)")
sql("LOADDATALOCALINPATH'examples/src/main/resources/kv1.txt'INTOTABLEsrc")
//QueriesareexpressedinHiveQL
sql("SELECT*FROMsrc").show()
//+---+-------+
//|key|value|
//+---+-------+
//|238|val_238|
//|86|val_86|
//|311|val_311|
//...
//Aggregationqueriesarealsosupported.
sql("SELECTCOUNT(*)FROMsrc").show()
//+--------+
//|count(1)|
//+--------+
//|500|
//+--------+
//TheresultsofSQLqueriesarethemselvesDataFramesandsupportallnormalfunctions.
valsqlDF=sql("SELECTkey,valueFROMsrcWHEREkey<10ORDERBYkey")
//TheitemsinDataFramesareoftypeRow,whichallowsyoutoaccesseachcolumnbyordinal.
valstringsDS=sqlDF.map{
caseRow(key:Int,value:String)=>s"Key:$key,Value:$value"
}
stringsDS.show()
//+--------------------+
//|value|
//+--------------------+
//|Key:0,Value:val_0|
//|Key:0,Value:val_0|
//|Key:0,Value:val_0|
//...
//YoucanalsouseDataFramestocreatetemporaryviewswithinaSparkSession.
valrecordsDF=spark.createDataFrame((1to100).map(i=>Record(i,s"val_$i")))
recordsDF.createOrReplaceTempView("records")
//QueriescanthenjoinDataFramedatawithdatastoredinHive.
sql("SELECT*FROMrecordsrJOINsrcsONr.key=s.key").show()
//+---+------+---+------+
//|key|value|key|value|
//+---+------+---+------+
//|2|val_2|2|val_2|
//|4|val_4|4|val_4|
//|5|val_5|5|val_5|
//...

内嵌Hive应用

如果要使用内嵌的Hive，什么都不用做，直接用就可以了。–conf:

spark.sql.warehouse.dir=

注意：如果你使用的是内部的Hive，在Spark2.0之后，spark.sql.warehouse.dir用于指定数据仓库的地址，如果你需要是用HDFS作为路径，那么需要将core-site.xml和hdfs-site.xml加入到Sparkconf目录，否则只会创建master节点上的warehouse目录，查询时会出现文件找不到的问题，这是需要向使用HDFS，则需要将metastore删除，重启集群。

外部Hive应用

如果想连接外部已经部署好的Hive，需要通过以下几个步骤。

a将Hive中的hive-site.xml拷贝或者软连接到Spark安装目录下的conf目录下。

b打开sparkshell，注意带上访问Hive元数据库的JDBC客户端。

$bin/spark-shell--masterspark://hadoop001:7077--jarsmysql-connector-java-5.1.27-bin.jar

JSON数据集

SparkSQL能够自动推测JSON数据集的结构，并将它加载为一个Dataset[Row].可以通过SparkSession.read.json()去加载一个Dataset[String]或者一个JSON文件.注意，这个JSON文件不是一个传统的JSON文件，每一行都得是一个JSON串。

{"name":"Michael"}
{"name":"Andy","age":30}
{"name":"Justin","age":19}
//Primitivetypes(Int,String,etc)andProducttypes(caseclasses)encodersare
//supportedbyimportingthiswhencreatingaDataset.
importspark.implicits._
//AJSONdatasetispointedtobypath.
//Thepathcanbeeitherasingletextfileoradirectorystoringtextfiles
valpath="examples/src/main/resources/people.json"
valpeopleDF=spark.read.json(path)
//TheinferredschemacanbevisualizedusingtheprintSchema()method
peopleDF.printSchema()
//root
//|--age:long(nullable=true)
//|--name:string(nullable=true)
//CreatesatemporaryviewusingtheDataFrame
peopleDF.createOrReplaceTempView("people")
//SQLstatementscanberunbyusingthesqlmethodsprovidedbyspark
valteenagerNamesDF=spark.sql("SELECTnameFROMpeopleWHEREageBETWEEN13AND19")
teenagerNamesDF.show()
//+------+
//|name|
//+------+
//|Justin|
//+------+
//Alternatively,aDataFramecanbecreatedforaJSONdatasetrepresentedby
//aDataset[String]storingoneJSONobjectperstring
valotherPeopleDataset=spark.createDataset(
"""{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}"""::Nil)
valotherPeople=spark.read.json(otherPeopleDataset)
otherPeople.show()
//+---------------+----+
//|address|name|
//+---------------+----+
//|[Columbus,Ohio]|Yin|
//+---------------+----+

JDBC

SparkSQL可以通过JDBC从关系型数据库中读取数据的方式创建DataFrame，通过对DataFrame一系列的计算后，还可以将数据再写回关系型数据库中。

注意，需要将相关的数据库驱动放到spark的类路径下。

$bin/spark-shell--masterspark://hadoop001:7077--jarsmysql-connector-java-5.1.27-bin.jar
//Note:JDBCloadingandsavingcanbeachievedviaeithertheload/saveorjdbcmethods
//LoadingdatafromaJDBCsource
valjdbcDF=spark.read.format("jdbc").option("url","jdbc:mysql://hadoop001:3306/rdd").option("dbtable","rddtable").option("user","root").option("password","hive").load()
valconnectionProperties=newProperties()
connectionProperties.put("user","root")
connectionProperties.put("password","hive")
valjdbcDF2=spark.read
.jdbc("jdbc:mysql://hadoop001:3306/rdd","rddtable",connectionProperties)
//SavingdatatoaJDBCsource
jdbcDF.write
.format("jdbc")
.option("url","jdbc:mysql://hadoop001:3306/rdd")
.option("dbtable","rddtable2")
.option("user","root")
.option("password","hive")
.save()
jdbcDF2.write
.jdbc("jdbc:mysql://hadoop001:3306/mysql","db",connectionProperties)
//Specifyingcreatetablecolumndatatypesonwrite
jdbcDF.write
.option("createTableColumnTypes","nameCHAR(64),commentsVARCHAR(1024)")
.jdbc("jdbc:mysql://hadoop001:3306/mysql","db",connectionProperties)

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。

声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。

Spark SQL常见4种数据源详解

热门推荐

随机推荐