SparkSQL之旅

1001,Gong Shaocheng,1
1002,Li Dachao,1
1003,Qiu Xin,1
1004,Cheng Jiangzhong,2
1005,Wo Binggang,3

[root@jfp3-1 spark-studio]# hdfs dfs -put employee.txt /user/spark_studio

[root@jfp3-1 spark-1.0.0-bin-hadoop2]# ./bin/spark-shell --master spark://192.168.0.71:7077

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext._

case class Employee(employeeId: Int, name: String, departmentId: Int)

// Create an RDD of Employee objects and register it as a table.
val employees = sc.textFile("hdfs://jfp3-1:8020/user/spark_studio/employee.txt").map(_.split(",")).map(p => Employee(p(0), p(1), p(2).trim.toInt))
employees.registerAsTable("employee")

// SQL statements can be run by using the sql methods provided by sqlContext.
val fsis = sql("SELECT name FROM employee WHERE departmentId = 1")

// The results of SQL queries are SchemaRDDs and support all the normal RDD operations.
// The columns of a row in the result can be accessed by ordinal.
fsis.map(t => "Name: " + t(0)).collect().foreach(println)

scala> val sqlContext = new org.apache.spark.sql.SQLContext(sc)
sqlContext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@17124319

scala> import sqlContext._
import sqlContext._

scala> case class Employee(employeeId: String, name: String, departmentId: Int)
defined class Employee

scala> val employees = sc.textFile("hdfs://jfp3-1:8020/user/spark_studio/employee.txt").map(_.split(",")).map(p => Employee(p(0), p(1), p(2).trim.toInt))
14/06/18 09:54:25 INFO MemoryStore: ensureFreeSpace(138763) called with curMem=0, maxMem=309225062
14/06/18 09:54:25 INFO MemoryStore: Block broadcast_0 stored as values to memory (estimated size 135.5 KB, free 294.8 MB)
employees: org.apache.spark.rdd.RDD[Employee] = MappedRDD[3] at map at <console>:19

scala> employees.registerAsTable("employee")

scala> val fsis = sql("SELECT name FROM employee WHERE departmentId = 1")
14/06/18 09:54:44 INFO Analyzer: Max iterations (2) reached for batch MultiInstanceRelations
14/06/18 09:54:44 INFO Analyzer: Max iterations (2) reached for batch CaseInsensitiveAttributeReferences
14/06/18 09:54:44 INFO SQLContext$$anon$1: Max iterations (2) reached for batch Add exchange
14/06/18 09:54:44 INFO SQLContext$$anon$1: Max iterations (2) reached for batch Prepare Expressions
fsis: org.apache.spark.sql.SchemaRDD = 
SchemaRDD[6] at RDD at SchemaRDD.scala:98
== Query Plan ==
Project [name#1:1]
 Filter (departmentId#2:2 = 1)
  ExistingRdd [employeeId#0,name#1,departmentId#2], MapPartitionsRDD[4] at mapPartitions at basicOperators.scala:174

scala> fsis.map(t => "Name: " + t(0)).collect().foreach(println)
14/06/18 09:55:27 INFO FileInputFormat: Total input paths to process : 1
14/06/18 09:55:27 INFO SparkContext: Starting job: collect at <console>:20
14/06/18 09:55:27 INFO DAGScheduler: Got job 0 (collect at <console>:20) with 2 output partitions (allowLocal=false)
14/06/18 09:55:27 INFO DAGScheduler: Final stage: Stage 0(collect at <console>:20)
14/06/18 09:55:27 INFO DAGScheduler: Parents of final stage: List()
14/06/18 09:55:27 INFO DAGScheduler: Missing parents: List()
14/06/18 09:55:27 INFO DAGScheduler: Submitting Stage 0 (MappedRDD[9] at map at <console>:20), which has no missing parents
14/06/18 09:55:27 INFO DAGScheduler: Submitting 2 missing tasks from Stage 0 (MappedRDD[9] at map at <console>:20)
14/06/18 09:55:27 INFO TaskSchedulerImpl: Adding task set 0.0 with 2 tasks
14/06/18 09:55:27 INFO TaskSetManager: Starting task 0.0:0 as TID 0 on executor 1: jfp3-2 (NODE_LOCAL)
14/06/18 09:55:27 INFO TaskSetManager: Serialized task 0.0:0 as 3508 bytes in 2 ms
14/06/18 09:55:27 INFO TaskSetManager: Starting task 0.0:1 as TID 1 on executor 2: jfp3-3 (NODE_LOCAL)
14/06/18 09:55:27 INFO TaskSetManager: Serialized task 0.0:1 as 3508 bytes in 0 ms
14/06/18 09:55:28 INFO TaskSetManager: Finished TID 1 in 1266 ms on jfp3-3 (progress: 1/2)
14/06/18 09:55:28 INFO TaskSetManager: Finished TID 0 in 1276 ms on jfp3-2 (progress: 2/2)
14/06/18 09:55:28 INFO DAGScheduler: Completed ResultTask(0, 1)
14/06/18 09:55:28 INFO DAGScheduler: Completed ResultTask(0, 0)
14/06/18 09:55:28 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool 
14/06/18 09:55:28 INFO DAGScheduler: Stage 0 (collect at <console>:20) finished in 1.284 s
14/06/18 09:55:28 INFO SparkContext: Job finished: collect at <console>:20, took 1.386154401 s

Name: Gong Shaocheng
Name: Li Dachao
Name: Qiu Xin

scala> val parquetFile = sqlContext.parquetFile("hdfs://jfp3-1:8020/user/spark_studio/employee.parquet")
14/06/18 10:24:36 INFO Analyzer: Max iterations (2) reached for batch MultiInstanceRelations
14/06/18 10:24:36 INFO Analyzer: Max iterations (2) reached for batch CaseInsensitiveAttributeReferences
14/06/18 10:24:36 INFO SQLContext$$anon$1: Max iterations (2) reached for batch Add exchange
14/06/18 10:24:36 INFO SQLContext$$anon$1: Max iterations (2) reached for batch Prepare Expressions
parquetFile: org.apache.spark.sql.SchemaRDD = 
SchemaRDD[13] at RDD at SchemaRDD.scala:98
== Query Plan ==
ParquetTableScan [employeeId#9,name#10,departmentId#11], (ParquetRelation hdfs://jfp3-1:8020/user/spark_studio/employee.parquet), None


scala> parquetFile.registerAsTable("parquetFile")


scala> val telcos = sql("SELECT name FROM parquetFile WHERE departmentId = 3")
14/06/18 10:24:37 INFO Analyzer: Max iterations (2) reached for batch MultiInstanceRelations
14/06/18 10:24:37 INFO Analyzer: Max iterations (2) reached for batch CaseInsensitiveAttributeReferences
14/06/18 10:24:37 INFO SQLContext$$anon$1: Max iterations (2) reached for batch Add exchange
14/06/18 10:24:37 INFO SQLContext$$anon$1: Max iterations (2) reached for batch Prepare Expressions
14/06/18 10:24:37 INFO MemoryStore: ensureFreeSpace(180579) called with curMem=138763, maxMem=309225062
14/06/18 10:24:37 INFO MemoryStore: Block broadcast_1 stored as values to memory (estimated size 176.3 KB, free 294.6 MB)
telcos: org.apache.spark.sql.SchemaRDD = 
SchemaRDD[14] at RDD at SchemaRDD.scala:98
== Query Plan ==
Project [name#10:0]
 Filter (departmentId#11:1 = 3)
  ParquetTableScan [name#10,departmentId#11], (ParquetRelation hdfs://jfp3-1:8020/user/spark_studio/employee.parquet), None

scala> telcos.collect().foreach(println)
14/06/18 10:24:40 INFO FileInputFormat: Total input paths to process : 2
14/06/18 10:24:40 INFO ParquetInputFormat: Total input paths to process : 2
14/06/18 10:24:40 INFO ParquetFileReader: reading summary file: hdfs://jfp3-1:8020/user/spark_studio/employee.parquet/_metadata
14/06/18 10:24:40 INFO deprecation: mapred.max.split.size is deprecated. Instead, use mapreduce.input.fileinputformat.split.maxsize
14/06/18 10:24:40 INFO deprecation: mapred.min.split.size is deprecated. Instead, use mapreduce.input.fileinputformat.split.minsize
14/06/18 10:24:40 INFO SparkContext: Starting job: collect at <console>:20
14/06/18 10:24:40 INFO DAGScheduler: Got job 2 (collect at <console>:20) with 2 output partitions (allowLocal=false)
14/06/18 10:24:40 INFO DAGScheduler: Final stage: Stage 2(collect at <console>:20)
14/06/18 10:24:40 INFO DAGScheduler: Parents of final stage: List()
14/06/18 10:24:40 INFO DAGScheduler: Missing parents: List()
14/06/18 10:24:40 INFO DAGScheduler: Submitting Stage 2 (SchemaRDD[14] at RDD at SchemaRDD.scala:98
== Query Plan ==
Project [name#10:0]
 Filter (departmentId#11:1 = 3)
  ParquetTableScan [name#10,departmentId#11], (ParquetRelation hdfs://jfp3-1:8020/user/spark_studio/employee.parquet), None), which has no missing parents
14/06/18 10:24:40 INFO DAGScheduler: Submitting 2 missing tasks from Stage 2 (SchemaRDD[14] at RDD at SchemaRDD.scala:98
== Query Plan ==
Project [name#10:0]
 Filter (departmentId#11:1 = 3)
  ParquetTableScan [name#10,departmentId#11], (ParquetRelation hdfs://jfp3-1:8020/user/spark_studio/employee.parquet), None)
14/06/18 10:24:40 INFO TaskSchedulerImpl: Adding task set 2.0 with 2 tasks
14/06/18 10:24:40 INFO TaskSetManager: Starting task 2.0:0 as TID 4 on executor 2: jfp3-3 (NODE_LOCAL)
14/06/18 10:24:40 INFO TaskSetManager: Serialized task 2.0:0 as 3116 bytes in 1 ms
14/06/18 10:24:40 INFO TaskSetManager: Starting task 2.0:1 as TID 5 on executor 0: jfp3-4 (NODE_LOCAL)
14/06/18 10:24:40 INFO TaskSetManager: Serialized task 2.0:1 as 3116 bytes in 1 ms
14/06/18 10:24:40 INFO DAGScheduler: Completed ResultTask(2, 0)
14/06/18 10:24:40 INFO TaskSetManager: Finished TID 4 in 200 ms on jfp3-3 (progress: 1/2)
14/06/18 10:24:42 INFO DAGScheduler: Completed ResultTask(2, 1)
14/06/18 10:24:42 INFO TaskSetManager: Finished TID 5 in 2162 ms on jfp3-4 (progress: 2/2)
14/06/18 10:24:42 INFO TaskSchedulerImpl: Removed TaskSet 2.0, whose tasks have all completed, from pool 
14/06/18 10:24:42 INFO DAGScheduler: Stage 2 (collect at <console>:20) finished in 2.177 s
14/06/18 10:24:42 INFO SparkContext: Job finished: collect at <console>:20, took 2.210887848 s

[Wo Binggang]

scala> all.collect().foreach(println)
14/06/18 10:37:45 INFO SparkContext: Starting job: collect at <console>:24
14/06/18 10:37:45 INFO DAGScheduler: Got job 6 (collect at <console>:24) with 2 output partitions (allowLocal=false)
14/06/18 10:37:45 INFO DAGScheduler: Final stage: Stage 6(collect at <console>:24)
14/06/18 10:37:45 INFO DAGScheduler: Parents of final stage: List()
14/06/18 10:37:45 INFO DAGScheduler: Missing parents: List()
14/06/18 10:37:45 INFO DAGScheduler: Submitting Stage 6 (SchemaRDD[33] at RDD at SchemaRDD.scala:98
== Query Plan ==
Project [name#19:1]
 Filter (departmentId#20:2 >= 1)
  ExistingRdd [employeeId#18,name#19,departmentId#20], MapPartitionsRDD[30] at mapPartitions at basicOperators.scala:174), which has no missing parents
14/06/18 10:37:45 INFO DAGScheduler: Submitting 2 missing tasks from Stage 6 (SchemaRDD[33] at RDD at SchemaRDD.scala:98
== Query Plan ==
Project [name#19:1]
 Filter (departmentId#20:2 >= 1)
  ExistingRdd [employeeId#18,name#19,departmentId#20], MapPartitionsRDD[30] at mapPartitions at basicOperators.scala:174)
14/06/18 10:37:45 INFO TaskSchedulerImpl: Adding task set 6.0 with 2 tasks
14/06/18 10:37:45 INFO TaskSetManager: Starting task 6.0:0 as TID 200 on executor 2: jfp3-3 (NODE_LOCAL)
14/06/18 10:37:45 INFO TaskSetManager: Serialized task 6.0:0 as 3541 bytes in 0 ms
14/06/18 10:37:45 INFO TaskSetManager: Starting task 6.0:1 as TID 201 on executor 1: jfp3-2 (NODE_LOCAL)
14/06/18 10:37:45 INFO TaskSetManager: Serialized task 6.0:1 as 3541 bytes in 1 ms
14/06/18 10:37:45 INFO TaskSetManager: Finished TID 200 in 33 ms on jfp3-3 (progress: 1/2)
14/06/18 10:37:45 INFO DAGScheduler: Completed ResultTask(6, 0)
14/06/18 10:37:45 INFO DAGScheduler: Completed ResultTask(6, 1)
14/06/18 10:37:45 INFO TaskSetManager: Finished TID 201 in 37 ms on jfp3-2 (progress: 2/2)
14/06/18 10:37:45 INFO TaskSchedulerImpl: Removed TaskSet 6.0, whose tasks have all completed, from pool 
14/06/18 10:37:45 INFO DAGScheduler: Stage 6 (collect at <console>:24) finished in 0.039 s
14/06/18 10:37:45 INFO SparkContext: Job finished: collect at <console>:24, took 0.052556716 s
[Gong Shaocheng]
[Li Dachao]
[Qiu Xin]
[Cheng Jiangzhong]
[Wo Binggang]