spark定制之六：sql版start.scala

上个版本的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
        """)
}

spark定制之六：sql版start.scala