15 友盟项目--资源文件工具类(ResourceUtil)、sql执行工具类(ExecSQLUtil）

/**
 * 资源文件工具类
 */
public class ResourceUtil {
    /**
     * 以String方式读取整个资源串
     */
    public static String readResourceAsString(String resource ,String charset) throws Exception {
        InputStream input = Thread.currentThread().getContextClassLoader().getResourceAsStream(resource) ;
        ByteArrayOutputStream baos = new ByteArrayOutputStream() ;

        byte[] buf = new byte[1024] ;
        int len = -1 ;
        while((len = input.read(buf)) != -1){
            baos.write(buf , 0 , len);
        }
        return new String(baos.toByteArray() , charset) ;
    }
    /**
     * 以String方式读取整个资源串
     */
    public static String readResourceAsString(String resource) throws Exception {
        InputStream input = Thread.currentThread().getContextClassLoader().getResourceAsStream(resource) ;
        ByteArrayOutputStream baos = new ByteArrayOutputStream() ;

        byte[] buf = new byte[1024] ;
        int len = -1 ;
        while((len = input.read(buf)) != -1){
            baos.write(buf , 0 , len);
        }
        String sql = new String(baos.toByteArray(), Charset.defaultCharset()) ;
        //替换掉注释
        sql = sql.replaceAll("--.*\r\n", "") ;
        return sql ;
    }

    /**
     * 将资源文件读取出来，形成list
     */
    public static List<String> readResourceAsList(String resource) throws Exception {
        List<String> list = new ArrayList<String>()  ;
        InputStream input = Thread.currentThread().getContextClassLoader().getResourceAsStream(resource);
        BufferedReader br = new BufferedReader(new InputStreamReader(input)) ;
        String line = null ;
        while((line = br.readLine()) != null){
            if(!line.trim().equals("")){
                list.add(line) ;
            }
        }
        return list ;
    }
}

package com.oldboy.umeng.spark.stat;

import com.oldboy.umeng.common.util.ResourceUtil;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * 执行脚本工具类
 */
public class ExecSQLUtil {
    /**
     * 执行sql脚本
     */
    public static void execSQLScript(SparkSession sess, String sqlScript) throws Exception {
        //资源工具类   把sql脚本转化为String
        String sqls = ResourceUtil.readResourceAsString(sqlScript);
        String arr[] = sqls.split(";");
        for (String sql : arr) {
            if (!sql.trim().equals("")) {
                sess.sql(sql).show(1000, false);
            }
        }
    }

    /**
     * 执行sqlsStr
     */
    public static void execSQLString(SparkSession sess, String sqlString) throws Exception {
        String arr[] = sqlString.split(";");
        for (String sql : arr) {
            if (!sql.trim().equals("")) {
                sess.sql(sql).show(1000, false);
            }
        }
    }
    /**
     * 执行sqlsStr
     */
    public static Dataset<Row> execSQLString2(SparkSession sess, String sqlString) throws Exception {
        String arr[] = sqlString.split(";");
        for (int i = 0 ; i< arr.length ; i ++) {
            if (!arr[i].trim().equals("")) {
                if(i != arr.length - 1){
                    sess.sql(arr[i]).show(); ;
                }
                else{
                    return sess.sql(arr[i]) ;
                }
            }
        }
        return null ;
    }

    /**
     * 注册函数
     */
    public static void execRegisterFuncs(SparkSession sess) throws Exception {
        execSQLScript(sess, "funcs.sql");
    }




}