第一種方式：Spark SQL可以將數據文件加載成RDD方式，然后將RDD轉成DataFrame或者DataSet。
第二種方式：從本地或者Cloud（hdfs hive S3等）

將文件加載成RDD

首先打開控制臺，輸入命令打開spark-shell：
./spark-shell --master local[2] --jars /home/iie4bu/software/mysql-connector-java-5.1.35.jar
然后加載本地文件：
val masterlog = sc.textFile("file:///home/iie4bu/app/spark-2.4.5-bin-2.6.0-cdh5.15.1/logs/spark-iie4bu-org.apache.spark.deploy.master.Master-1-manager.out.2")

此時已經將本地文件加載成RDD了，還無法使用SQL進行查詢，因為沒有轉成DataFrame。

將RDD轉成DataFrame

import org.apache.spark.sql.Row val masterRDD = masterlog.map(x => Row(x))import org.apache.spark.sql.types._ val schemaString = "line"val fields = schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, nullable = true))val schema = StructType(fields)val masterDF = spark.createDataFrame(masterRDD, schema) masterDF.printSchema masterDF.show

這樣就可以使用DataFrame中的相關查詢了。也可以將DataFrame注冊成一張表。

將DataFrame注冊成表

masterDF.createOrReplaceTempView("master_logs") spark.sql("select * from master_logs limit 10").show(false)

如果使用JSON或者Parquet，schema信息不需要手寫，spark可以進行推測

使用Spark導入外部數據源，這種方式不需要手動寫schema了：

val usersDF = spark.read.format("parquet").load("file:///home/iie4bu/app/spark-2.4.5-bin-2.6.0-cdh5.15.1/examples/src/main/resources/users.parquet") usersDF.printSchema

生成DataFrame之后，也可以使用創建表來執行sql：

usersDF.createOrReplaceTempView("users_info") spark.sql("select * from users_info where name = 'Ben'").show

更簡單的實現方式

spark.sql("select * from parquet.`file:///home/iie4bu/app/spark-2.4.5-bin-2.6.0-cdh5.15.1/examples/src/main/resources/users.parquet` where name = 'Ben'").show

讀取hdfs或者s3數據

讀取成RDD：

val hdfsRDD = sc.textFile("hdfs://path/file") val s3RDD = sc.textFile("s3a://bucket/object")

直接讀取成DataFrame：

spark.read.format("text").load("hdfs://path/file") spark.read.format("text").load("s3a://bucket/object")

總結

以上是生活随笔為你收集整理的Spark SQL 加载数据的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。