這篇文章主要介紹Spark SQL中常見4種數(shù)據(jù)源是什么,文中介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們一定要看完!
成都創(chuàng)新互聯(lián)公司2013年至今,先為上栗等服務(wù)建站,上栗等地企業(yè),進(jìn)行企業(yè)商務(wù)咨詢服務(wù)。為上栗企業(yè)網(wǎng)站制作PC+手機(jī)+微官網(wǎng)三網(wǎng)同步一站式服務(wù)解決您的所有建站問(wèn)題。
通用load/write方法
手動(dòng)指定選項(xiàng)
Spark SQL的DataFrame接口支持多種數(shù)據(jù)源的操作。一個(gè)DataFrame可以進(jìn)行RDDs方式的操作,也可以被注冊(cè)為臨時(shí)表。把DataFrame注冊(cè)為臨時(shí)表之后,就可以對(duì)該DataFrame執(zhí)行SQL查詢。
Spark SQL的默認(rèn)數(shù)據(jù)源為Parquet格式。數(shù)據(jù)源為Parquet文件時(shí),Spark SQL可以方便的執(zhí)行所有的操作。
修改配置項(xiàng)spark.sql.sources.default,可修改默認(rèn)數(shù)據(jù)源格式。
scala> val df = 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")
當(dāng)數(shù)據(jù)源格式不是parquet格式文件時(shí),需要手動(dòng)指定數(shù)據(jù)源的格式。數(shù)據(jù)源格式需要指定全名(例如:org.apache.spark.sql.parquet),如果數(shù)據(jù)源格式為內(nèi)置格式,則只需要指定簡(jiǎn)稱json, parquet, jdbc, orc, libsvm, csv, text來(lái)指定數(shù)據(jù)的格式。
可以通過(guò)SparkSession提供的read.load方法用于通用加載數(shù)據(jù),使用write和save保存數(shù)據(jù)。
scala> val peopleDF = 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>
除此之外,可以直接運(yùn)行SQL在文件上:
val sqlDF = spark.sql("SELECT * FROM parquet.`hdfs://hadoop001:9000/namesAndAges.parquet`") sqlDF.show()
文件保存選項(xiàng)
可以采用SaveMode執(zhí)行存儲(chǔ)操作,SaveMode定義了對(duì)數(shù)據(jù)的處理模式。需要注意的是,這些保存模式不使用任何鎖定,不是原子操作。此外,當(dāng)使用Overwrite方式執(zhí)行時(shí),在輸出新數(shù)據(jù)之前原數(shù)據(jù)就已經(jīng)被刪除。SaveMode詳細(xì)介紹如下表:
Scala/Java | Any Language | Meaning |
---|---|---|
SaveMode.ErrorIfExists(default) | “error”(default) | 如果文件存在,則報(bào)錯(cuò) |
SaveMode.Append | “append” | 追加 |
SaveMode.Overwrite | “overwrite” | 覆寫 |
SaveMode.Ignore | “ignore” | 數(shù)據(jù)存在,則忽略 |
Parquet文件
Parquet讀寫
Parquet格式經(jīng)常在Hadoop生態(tài)圈中被使用,它也支持Spark SQL的全部數(shù)據(jù)類型。Spark SQL 提供了直接讀取和存儲(chǔ) Parquet 格式文件的方法。
// Encoders for most common types are automatically provided by importing spark.implicits._ import spark.implicits._ val peopleDF = spark.read.json("examples/src/main/resources/people.json") // DataFrames can be saved as Parquet files, maintaining the schema information peopleDF.write.parquet("hdfs://hadoop001:9000/people.parquet") // Read in the parquet file created above // Parquet files are self-describing so the schema is preserved // The result of loading a Parquet file is also a DataFrame val parquetFileDF = spark.read.parquet("hdfs://hadoop001:9000/people.parquet") // Parquet files can also be used to create a temporary view and then used in SQL statements parquetFileDF.createOrReplaceTempView("parquetFile") val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19") namesDF.map(attributes => "Name: " + attributes(0)).show() // +------------+ // | value| // +------------+ // |Name: Justin| // +------------+
解析分區(qū)信息
對(duì)表進(jìn)行分區(qū)是對(duì)數(shù)據(jù)進(jìn)行優(yōu)化的方式之一。在分區(qū)的表內(nèi),數(shù)據(jù)通過(guò)分區(qū)列將數(shù)據(jù)存儲(chǔ)在不同的目錄下。Parquet數(shù)據(jù)源現(xiàn)在能夠自動(dòng)發(fā)現(xiàn)并解析分區(qū)信息。例如,對(duì)人口數(shù)據(jù)進(jìn)行分區(qū)存儲(chǔ),分區(qū)列為gender和country,使用下面的目錄結(jié)構(gòu):
path └── to └── table ├── gender=male │ ├── ... │ │ │ ├── country=US │ │ └── data.parquet │ ├── country=CN │ │ └── data.parquet │ └── ... └── gender=female ├── ... │ ├── country=US │ └── data.parquet ├── country=CN │ └── data.parquet └── ...
通過(guò)傳遞path/to/table給 SQLContext.read.parque
或SQLContext.read.load,Spark SQL將自動(dòng)解析分區(qū)信息。
返回的DataFrame的Schema如下:
root |-- name: string (nullable = true) |-- age: long (nullable = true) |-- gender: string (nullable = true) |-- country: string (nullable = true)
需要注意的是,數(shù)據(jù)的分區(qū)列的數(shù)據(jù)類型是自動(dòng)解析的。當(dāng)前,支持?jǐn)?shù)值類型和字符串類型。自動(dòng)解析分區(qū)類型的參數(shù)為:
spark.sql.sources.partitionColumnTypeInference.enabled
默認(rèn)值為true。
如果想關(guān)閉該功能,直接將該參數(shù)設(shè)置為disabled。此時(shí),分區(qū)列數(shù)據(jù)格式將被默認(rèn)設(shè)置為string類型,不再進(jìn)行類型解析。
Schema合并
像ProtocolBuffer、Avro和Thrift那樣,Parquet也支持Schema evolution(Schema演變)。用戶可以先定義一個(gè)簡(jiǎn)單的Schema,然后逐漸的向Schema中增加列描述。通過(guò)這種方式,用戶可以獲取多個(gè)有不同Schema但相互兼容的Parquet文件?,F(xiàn)在Parquet數(shù)據(jù)源能自動(dòng)檢測(cè)這種情況,并合并這些文件的schemas。
因?yàn)镾chema合并是一個(gè)高消耗的操作,在大多數(shù)情況下并不需要,所以Spark SQL從1.5.0開始默認(rèn)關(guān)閉了該功能??梢酝ㄟ^(guò)下面兩種方式開啟該功能:
當(dāng)數(shù)據(jù)源為Parquet文件時(shí),將數(shù)據(jù)源選項(xiàng)mergeSchema設(shè)置為true。
設(shè)置全局SQL選項(xiàng):
spark.sql.parquet.mergeSchema為true。
// sqlContext from the previous example is used in this example. // This is used to implicitly convert an RDD to a DataFrame. import spark.implicits._ // Create a simple DataFrame, stored into a partition directory val df1 = sc.makeRDD(1 to 5).map(i => (i, i * 2)).toDF("single", "double") df1.write.parquet("hdfs://hadoop001:9000/data/test_table/key=1") // Create another DataFrame in a new partition directory, // adding a new column and dropping an existing column val df2 = sc.makeRDD(6 to 10).map(i => (i, i * 3)).toDF("single", "triple") df2.write.parquet("hdfs://hadoop001:9000/data/test_table/key=2") // Read the partitioned table val df3 = spark.read.option("mergeSchema", "true").parquet("hdfs://hadoop001:9000/data/test_table") df3.printSchema() // The final schema consists of all 3 columns in the Parquet files together // with the partitioning column appeared in the partition directory paths. // root // |-- single: int (nullable = true) // |-- double: int (nullable = true) // |-- triple: int (nullable = true) // |-- key : int (nullable = true)
Hive數(shù)據(jù)源
Apache Hive是Hadoop上的SQL引擎,Spark SQL編譯時(shí)可以包含Hive支持,也可以不包含。包含Hive支持的Spark SQL可以支持Hive表訪問(wèn)、UDF(用戶自定義函數(shù))以及 Hive 查詢語(yǔ)言(HiveQL/HQL)等。需要強(qiáng)調(diào)的 一點(diǎn)是,如果要在Spark SQL中包含Hive的庫(kù),并不需要事先安裝Hive。一般來(lái)說(shuō),最好還是在編譯Spark SQL時(shí)引入Hive支持,這樣就可以使用這些特性了。如果你下載的是二進(jìn)制版本的 Spark,它應(yīng)該已經(jīng)在編譯時(shí)添加了 Hive 支持。
若要把Spark SQL連接到一個(gè)部署好的Hive上,你必須把hive-site.xml復(fù)制到 Spark的配置文件目錄中($SPARK_HOME/conf)。即使沒(méi)有部署好Hive,Spark SQL也可以運(yùn)行。
需要注意的是,如果你沒(méi)有部署好Hive,Spark SQL會(huì)在當(dāng)前的工作目錄中創(chuàng)建出自己的Hive 元數(shù)據(jù)倉(cāng)庫(kù),叫作 metastore_db。此外,如果你嘗試使用 HiveQL 中的 CREATE TABLE (并非 CREATE EXTERNAL TABLE)語(yǔ)句來(lái)創(chuàng)建表,這些表會(huì)被放在你默認(rèn)的文件系統(tǒng)中的 /user/hive/warehouse 目錄中(如果你的 classpath 中有配好的 hdfs-site.xml,默認(rèn)的文件系統(tǒng)就是 HDFS,否則就是本地文件系統(tǒng))。
import java.io.File import org.apache.spark.sql.Row import org.apache.spark.sql.SparkSession case class Record(key: Int, value: String) // warehouseLocation points to the default location for managed databases and tables val warehouseLocation = new File("spark-warehouse").getAbsolutePath val spark = SparkSession .builder() .appName("Spark Hive Example") .config("spark.sql.warehouse.dir", warehouseLocation) .enableHiveSupport() .getOrCreate() import spark.implicits._ import spark.sql sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)") sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src") // Queries are expressed in HiveQL sql("SELECT * FROM src").show() // +---+-------+ // |key| value| // +---+-------+ // |238|val_238| // | 86| val_86| // |311|val_311| // ... // Aggregation queries are also supported. sql("SELECT COUNT(*) FROM src").show() // +--------+ // |count(1)| // +--------+ // | 500 | // +--------+ // The results of SQL queries are themselves DataFrames and support all normal functions. val sqlDF = sql("SELECT key, value FROM src WHERE key < 10 ORDER BY key") // The items in DataFrames are of type Row, which allows you to access each column by ordinal. val stringsDS = sqlDF.map { case Row(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| // ... // You can also use DataFrames to create temporary views within a SparkSession. val recordsDF = spark.createDataFrame((1 to 100).map(i => Record(i, s"val_$i"))) recordsDF.createOrReplaceTempView("records") // Queries can then join DataFrame data with data stored in Hive. sql("SELECT * FROM records r JOIN src s ON r.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| // ...
內(nèi)嵌Hive應(yīng)用
如果要使用內(nèi)嵌的Hive,什么都不用做,直接用就可以了。 –conf :
spark.sql.warehouse.dir=
注意:如果你使用的是內(nèi)部的Hive,在Spark2.0之后,spark.sql.warehouse.dir用于指定數(shù)據(jù)倉(cāng)庫(kù)的地址,如果你需要是用HDFS作為路徑,那么需要將core-site.xml和hdfs-site.xml 加入到Spark conf目錄,否則只會(huì)創(chuàng)建master節(jié)點(diǎn)上的warehouse目錄,查詢時(shí)會(huì)出現(xiàn)文件找不到的問(wèn)題,這是需要向使用HDFS,則需要將metastore刪除,重啟集群。
外部Hive應(yīng)用
如果想連接外部已經(jīng)部署好的Hive,需要通過(guò)以下幾個(gè)步驟。
a 將Hive中的hive-site.xml拷貝或者軟連接到Spark安裝目錄下的conf目錄下。
b 打開spark shell,注意帶上訪問(wèn)Hive元數(shù)據(jù)庫(kù)的JDBC客戶端。
$ bin/spark-shell --master spark://hadoop001:7077 --jars MySQL-connector-java-5.1.27-bin.jar
JSON數(shù)據(jù)集
Spark SQL 能夠自動(dòng)推測(cè) JSON數(shù)據(jù)集的結(jié)構(gòu),并將它加載為一個(gè)Dataset[Row]. 可以通過(guò)SparkSession.read.json()去加載一個(gè) Dataset[String]或者一個(gè)JSON 文件.注意,這個(gè)JSON文件不是一個(gè)傳統(tǒng)的JSON文件,每一行都得是一個(gè)JSON串。
{"name":"Michael"} {"name":"Andy", "age":30} {"name":"Justin", "age":19} // Primitive types (Int, String, etc) and Product types (case classes) encoders are // supported by importing this when creating a Dataset. import spark.implicits._ // A JSON dataset is pointed to by path. // The path can be either a single text file or a directory storing text files val path = "examples/src/main/resources/people.json" val peopleDF = spark.read.json(path) // The inferred schema can be visualized using the printSchema() method peopleDF.printSchema() // root // |-- age: long (nullable = true) // |-- name: string (nullable = true) // Creates a temporary view using the DataFrame peopleDF.createOrReplaceTempView("people") // SQL statements can be run by using the sql methods provided by spark val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19") teenagerNamesDF.show() // +------+ // | name| // +------+ // |Justin| // +------+ // Alternatively, a DataFrame can be created for a JSON dataset represented by // a Dataset[String] storing one JSON object per string val otherPeopleDataset = spark.createDataset( """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil) val otherPeople = spark.read.json(otherPeopleDataset) otherPeople.show() // +---------------+----+ // | address|name| // +---------------+----+ // |[Columbus,Ohio]| Yin| // +---------------+----+
JDBC
Spark SQL可以通過(guò)JDBC從關(guān)系型數(shù)據(jù)庫(kù)中讀取數(shù)據(jù)的方式創(chuàng)建DataFrame,通過(guò)對(duì)DataFrame一系列的計(jì)算后,還可以將數(shù)據(jù)再寫回關(guān)系型數(shù)據(jù)庫(kù)中。
注意,需要將相關(guān)的數(shù)據(jù)庫(kù)驅(qū)動(dòng)放到spark的類路徑下。
$ bin/spark-shell --master spark://hadoop001:7077 --jars mysql-connector-java-5.1.27-bin.jar // Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods // Loading data from a JDBC source val jdbcDF = spark.read.format("jdbc").option("url", "jdbc:mysql://hadoop001:3306/rdd").option("dbtable", " rddtable").option("user", "root").option("password", "hive").load() val connectionProperties = new Properties() connectionProperties.put("user", "root") connectionProperties.put("password", "hive") val jdbcDF2 = spark.read .jdbc("jdbc:mysql://hadoop001:3306/rdd", "rddtable", connectionProperties) // Saving data to a JDBC source 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) // Specifying create table column data types on write jdbcDF.write .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)") .jdbc("jdbc:mysql://hadoop001:3306/mysql", "db", connectionProperties)
以上是“Spark SQL中常見4種數(shù)據(jù)源是什么”這篇文章的所有內(nèi)容,感謝各位的閱讀!希望分享的內(nèi)容對(duì)大家有幫助,更多相關(guān)知識(shí),歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道!
新聞名稱:SparkSQL中常見4種數(shù)據(jù)源是什么
文章URL:http://m.rwnh.cn/article12/ippodc.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供網(wǎng)站營(yíng)銷、做網(wǎng)站、定制網(wǎng)站、關(guān)鍵詞優(yōu)化、網(wǎng)站排名、
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請(qǐng)盡快告知,我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如需處理請(qǐng)聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時(shí)需注明來(lái)源: 創(chuàng)新互聯(lián)