上个版本的start.scala用的是HiveContext，这个是SQLContext的，不需编译。
# cat testperson.txt #字段用table键分隔
zs 10 30.0
li 12 32.0
# spark-shell -i:start.scala
scala> help
根据提示逐步运行
import org.apache.spark.sql.SchemaRDD var FIELD_SEPERATOR = "\t" var RECORD_SEPERATOR = "\n" var lastrdd : SchemaRDD = null object MyFileUtil extends java.io.Serializable { import org.apache.hadoop.fs.Path import org.apache.hadoop.fs.FileSystem import org.apache.hadoop.fs.FileStatus import scala.collection.mutable.ListBuffer def regularFile(filepath:String):String = { if(filepath == "") { filepath; } else if(filepath.startsWith("hdfs:")) { filepath } else if(filepath.startsWith("file:")) { filepath } else if(filepath.startsWith("/")) { "file://" + filepath } else { val workdir = System.getProperty("user.dir") "file://" + workdir + "/" + filepath } } var SAFEMINPATH_LENGTH : Int = 24 def getFileSystem(filepath:String) = { if(filepath.startsWith("hdfs:")) { FileSystem.get(new org.apache.hadoop.conf.Configuration()); } else if(filepath.startsWith("file:")) { FileSystem.getLocal(new org.apache.hadoop.conf.Configuration()); } else { throw new Exception("file path invalid") } } def deletePath(filepath:String) = { if(filepath.length < SAFEMINPATH_LENGTH) throw new Exception("file path is to short") var fs : FileSystem = getFileSystem(filepath) if (fs.exists(new Path(filepath))) { fs.delete(new Path(filepath), true); } } def listFile(fs:FileSystem, path:Path, pathlist:ListBuffer[Path], statuslist:ListBuffer[FileStatus]=null) { if ( fs.exists(path) ) { val substatuslist = fs.listStatus(path); for(substatus <- substatuslist){ if(statuslist != null) statuslist.append(substatus) if(substatus.isDir()){ listFile(fs,substatus.getPath(),pathlist); }else{ pathlist.append(substatus.getPath()); } } } } def hasContext(filepath:String) = { val realpath = regularFile(filepath) val fs = getFileSystem(realpath) val pathlist = ListBuffer[Path]() val statuslist = ListBuffer[FileStatus]() listFile(fs,new Path(filepath),pathlist,statuslist) var length:Long = 0 for( status <- statuslist ) length += status.getLen() length > 0 } } org.apache.spark.repl.Main.interp.command(""" class MySchemaRDD(rdd:org.apache.spark.sql.SchemaRDD) extends java.io.Serializable { def go() = { var startstr = "" var endstr = RECORD_SEPERATOR val result = rdd.collect result.foreach( x => print(x.mkString(startstr,FIELD_SEPERATOR,endstr)) ) } def result() = { rdd.collect } def saveto(output: String) = { import org.apache.hadoop.io.{NullWritable,Text} var startstr = "" var endstr = RECORD_SEPERATOR if(output.startsWith("hdfs:")) { val outputpath = MyFileUtil.regularFile(output) MyFileUtil.deletePath(outputpath) rdd.map(x => (NullWritable.get(), new Text(x.mkString(FIELD_SEPERATOR))) ).saveAsHadoopFile[ org.apache.hadoop.mapred.TextOutputFormat[NullWritable, Text] ](outputpath) } else { val outputpath = MyFileUtil.regularFile(output) MyFileUtil.deletePath(outputpath) val result = rdd.collect() val writer = new java.io.FileWriter(output) result.foreach(x => writer.write(x.mkString(startstr,FIELD_SEPERATOR,endstr)) ) writer.close() } } } object MySchemaRDD { implicit def toMySchemaRDD(rdd:org.apache.spark.sql.SchemaRDD) = new MySchemaRDD(rdd) } """) val ssc = new org.apache.spark.sql.SQLContext(sc) import ssc._ import MySchemaRDD._ def getRegisterString(rddname:String,classname:String,tablename:String,tabledef:String) : String = { val members = tabledef.trim.split(",").map(_.trim.split(" ").filter(""!=)).map(x => (x(0).trim,x(1).trim.head.toString.toUpperCase+x(1).trim.tail)) val classmemberdef = members.map(x => (x._1+":"+x._2)).mkString(",") val convertstr = members.map(x => x._2).zipWithIndex.map(x => "t("+x._2+").to"+x._1).mkString(",") return s""" case class ${classname}(${classmemberdef}) val schemardd = ${rddname}.map(_.split("${FIELD_SEPERATOR}")).map(t=>${classname}(${convertstr})) ssc.registerRDDAsTable(schemardd,"${tablename}") """ } org.apache.spark.repl.Main.interp.command(""" class MyCommandTranslator(cmd:String) extends java.io.Serializable { def go()(implicit f: SchemaRDD => MySchemaRDD) = { lastrdd = sql(cmd) lastrdd.go() } def saveto(output: String)(implicit f: SchemaRDD => MySchemaRDD) = { lastrdd = sql(cmd) lastrdd.saveto(output) } def result()(implicit f: SchemaRDD => MySchemaRDD) = { lastrdd = sql(cmd) lastrdd.result() } // def hqlgo()(implicit f: SchemaRDD => MySchemaRDD) = { // lastrdd = hql(cmd) // lastrdd.go() // } // // def hqlsaveto(output: String)(implicit f: SchemaRDD => MySchemaRDD) = { // lastrdd = hql(cmd) // lastrdd.saveto(output) // } // // def hqlresult()(implicit f: SchemaRDD => MySchemaRDD) = { // lastrdd = hql(cmd) // lastrdd.result() // } def defineas(tabledef:String) = { if( tabledef != "" ) { org.apache.spark.repl.Main.interp.command( getRegisterString(cmd,cmd.toUpperCase,cmd,tabledef) ) } else { org.apache.spark.repl.Main.interp.command( "ssc.registerRDDAsTable(${cmd},\"${cmd}\")" ) } } def from(filepath:String) { if( cmd.trim.startsWith("create table ") ) { val tablename = cmd.trim.substring(13).trim().split(" ")(0) val leftstr = cmd.substring(13).trim().substring(tablename.length).trim() val tabledef = leftstr.substring(1,leftstr.length-1).trim() val realfile = MyFileUtil.regularFile(filepath) org.apache.spark.repl.Main.interp.command( "val "+tablename+" = sc.textFile(\""+realfile+"\")" ) new MyCommandTranslator(tablename).defineas(tabledef) } else { println("usage:") println("\"create table sometablename (field1 string,field2 int...)\" from \"somefile or hdfs:somepath\"") } } def isok() = { if(cmd.contains(".") || cmd.contains("/")) { MyFileUtil.hasContext(cmd) } else { val res = sql(s"select count(*) from ${cmd}").result() val count = res(0).getLong(0) count > 0 } } } object MyCommandTranslator { implicit def stringToTranslator(cmd:String) = new MyCommandTranslator(cmd) def show(tabledata:Array[org.apache.spark.sql.Row]) = { tabledata.foreach( x => println(x.mkString("\t"))) } } """) def to = MyCommandTranslator import MyCommandTranslator._ val onetable = sql("select 1 as id") ssc.registerRDDAsTable(onetable,"onetable") def help = { println("""example: "create table testperson (name string,age int,weight double)" from "testperson.txt" "select * from testperson" go "select * from testperson" saveto "somelocalfile.txt" "select * from testperson" saveto "hdfs:/basedir/parentdir/testperson" "testperson" isok "somelocalfile.txt" isok "hdfs:/basedir/parentdir/testperson" isok val data = "select * from testperson" result to show data val somerdddata = sc.textFile("hdfs:/basedir/parentdir/testperson")<span style="font-family: Arial, Helvetica, sans-serif;"> </span> "somerdddata" defineas "name string,age int,weight double" "select * from somerdddata" go if you want to see the help of enveronment, please type :help """) }