SparkSQL项目中的应用

Spark是一个通用的大规模数据快速处理引擎。可以简单理解为Spark就是一个大数据分布式处理框架。基于内存计算的Spark的计算速度要比Hadoop的MapReduce快上100倍以上，基于磁盘的计算速度也快于10倍以上。Spark运行在Hadoop第二代的yarn集群管理之上，可以轻松读取Hadoop的任何数据。能够读取HBase、HDFS等Hadoop的数据源。

    从Spark 1.0版本起，Spark开始支持Spark SQL，它最主要的用途之一就是能够直接从Spark平台上面获取数据。并且Spark SQL提供比较流行的Parquet列式存储格式以及从Hive表中直接读取数据的支持。之后，Spark SQL还增加了对JSON等其他格式的支持。到了Spark 1.3 版本Spark还可以使用SQL的方式进行DataFrames的操作。我们通过JDBC的方式通过前台业务逻辑执行相关sql的增删改查，通过远程连接linux对文件进行导入处理，使项目能够初步支持Spark平台，现如今已支持Spark1.4版本。

SparkSQL具有内置的SQL扩展的基类实现Catalyst，提供了提供了解析（一个非常简单的用Scala语言编写的SQL解析器）、执行（Spark Planner,生成基于RDD的物理计划）和绑定（数据完全存放于内存中）。

前台我们使用ThriftServer连接后台SparkSQL,它是一个JDBC/ODBC接口，通过配置Hive-site.xml，就可以使前台用JDBC/ODBC连接ThriftServer来访问SparkSQL的数据。ThriftServer通过调用hive元数据信息找到表或文件信息在hdfs上的具体位置，并通过Spark的RDD实现了hive的接口。对于标签、客户群探索的增、删、改、查都是通过SparkSQL对HDFS上存储的相应表文件进行操作，突破了传统数据库的瓶颈，同时为以后的客户群智能分析作了铺垫。

1.数据的存储格式

我们使用Parquet面向列存存储的文件存储结构，因为Parquet具有高压缩比的特点且适合嵌套数据类型的存储，能够避免不必要的IO性能。Parquet建表如下所示:

CREATE TABLE dw_coclbl_d01_20140512_lzo_256_parquet(op_time string,

join_id double, city_id int, product_no string, brand_id int, vip_level int, county_id int, l2_01_01_04 double,  l2_01_01_04_01 double)

ROW FORMAT SERDE 'parquet.hive.serde.ParquetHiveSerDe'STORED AS

INPUTFORMAT 'parquet.hive.DeprecatedParquetInputFormat'

OUTPUTFORMAT 'parquet.hive.DeprecatedParquetOutputFormat';

2、由于压缩文件占用的空间较少，文件load的速度比较快。故使用压缩文件进行数据的load.使用gzip进行压缩时，单个文件只能在一个节点上进行load,加载时间很长。使用split命令将解压后的csv文件分割成多个256M的小文件，机器上每个block块的大小为128M,故将小文件分割为128M或256M以保证效率。由于Parquet存储格式暂时只支持Gzip，项目中暂时使用Gzip压缩格式。通过在控制台输入set mapreduce.output.fileoutputformat.compress=true指令命令设置压缩格式为true。再执行set mapreduce.output. fileoutput format.compress.codec = org.apache.hadoop.io.compress.GzipCodec;将文件的压缩格式设置为Gzip压缩格式

3、数据的导入。使用的是Apache的一个项目，最早作为Hadoop的一个第三方模块存在，主要功能是在Hadoop(hive)与传统的数据库(mysql、oracle等)间进行数据的传递，可以将一个关系型数据库中的数据导入到Hadoop的HDFS中，也可以将HDFS的数据导进到关系数据库中。

由于执行sqoop导入需要通过yarn的任务调度进行mapreduce，由于spark开启后即便在空闲状态下也不释放内存，故修改spark-env.sh配置，分配多余内存以便sqoop执行。

Create job -f 3 -t 4

Creating job for links with from id 3 and to id 4

    Please fill following values to create new job object

    Name: Sqoopy

    From database configuration

    Schema name: hive

    Table name: TBLS

    Table SQL statement: 

    Table column names: 

    Partition column name: 

    Null value allowed for the partition column: 

    Boundary query: 

    ToJob configuration

    Output format: 

      0 : TEXT_FILE

      1 : SEQUENCE_FILE

    Choose: 0

    Compression format: 

successfully created with validation status OK  and persistent id 2

     0 : NONE

      1 : DEFAULT

      2 : DEFLATE

      3 : GZIP

      4 : BZIP2

      5 : LZO

      6 : LZ4

      7 : SNAPPY

      8 : CUSTOM

    Choose: 0

    Custom compression format: 

    Output directory: hdfs://hadoop000:8020/sqoop2

    Throttling resources

    Extractors: 

    Loaders: 

    New job was

4、前台与后台交互工具类

工具类提供静态的方法，可以进行相应业务逻辑的调用，由于Hadoop集群存在于服务器端，前台需要实现跨平台服务器的连接，才能执行相应的Hadoop命令，实现对HDFS上文件的操作。此次设计的ShellUtils类，通过jsch连接Linux服务器执行shell命令.

private static JSch jsch;  

    private static Session session;  

    public static void connect(String user, String passwd, String host) throws JSchException {  

        jsch = new JSch();  

        session = jsch.getSession(user, host,22);  

        session.setPassword(passwd);  

        java.util.Properties config = new java.util.Properties();  

        config.put("StrictHostKeyChecking", "no");  

        session.setConfig(config);  

通过传入的Linux命令、用户名、密码等参数对远程linux服务器进行连接。由于执行Hadoop命令根据不同文件的大小所需占用的时间是不同的，在hadoop尚未将文件完全从hdfs上合并到本地时，本地会提前生成文件但文件内容为空，至此这里需要多传入前台客户群探索出来的客户群数目与文件条数进行对比，倘若数目相同则说明执行完毕。

 CodecUtil类，用来实现不同类型压缩文件的解压工作，通过传入的压缩类型，利用反射机制锁定压缩的类型，由于存储在hdfs上的文件都是以文件块的形式存在的，所以首先需要获取hdfs中文件的二级子目录，遍历查询到每一个文件块的文件路径，随后通过输入输出流进行文件的解压工作。然后将此类打包成jar包放入集群中，通过前台远程连接服务端，执行hadoop命令操作执行，实现类部分代码如下：

public class CodecUtil{

public static void main(String[] args) throws Exception {

        //compress("org.apache.hadoop.io.compress.GzipCodec");

              String listName = args[0];

              String codecType = args[1];

              String hdfsPath = args[2];

        uncompress(listName,codecType,hdfsPath);

//解压缩

public static void uncompress(String listName,String CodecType,String hdfsPath) throws Exception{

        Class<?> codecClass = Class.forName(CodecType);

        Configuration conf = new Configuration();

        FileSystem fs = FileSystem.get(conf);

        Path listf =new Path(hdfsPath+listName);

      //获取根目录下的所有2级子文件目录

        FileStatus stats[]=fs.listStatus(listf);

        CompressionCodec codec = (CompressionCodec)

ReflectionUtils.newInstance(codecClass, conf);

        int i;

        for ( i = 0; i < stats.length; i++){

               //获得子文件块的文件路径

            String Random = findRandom();

               Path list = new Path(stats[i].getPath().toString());

  InputStream in = codec.createInputStream(inputStream);

  FSDataOutputStream output = fs.create(new Path(hdfsPath + listName+"/"+unListName));

            IOUtils.copyBytes(in, output, conf);

            IOUtils.closeStream(in);

        }

    }

}

5、导入生成客户群

由于sparksql不支持insert into value语句，无法通过jdbc方式连接后台HDFS通过sparksql对文件进行导入数据的操作。于是将需要导入的csv文件通过ftp方式上传到远程服务器，再将文件通过load的方式导入表中，实现导入生成客户群的功能。

// 将文件上传到ftp服务器

       CiFtpInfo ftp = customFileRelService.getCiFtpInfoByFtpType(1);

       FtpUtil.ftp(ftp.getFtpServerIp(),ftp.getFtpPort(),ftp.getFtpUser(),DES.decrypt

(ftp.getFtpPwd()), ftpFileName, ftp.getFtpPath());

// 将文件load到表中

              String ftpPath = ftp.getFtpPath();

              if (!ftpPath.endsWith("/")) {

                     ftpPath = ftpPath + "/";

              }

String sql = " LOAD DATA LOCAL INPATH '" + ftpPath + fileName

       + "' OVERWRITE INTO TABLE " + tabName;

              log.info("loadSql=" + sql);

customersService.executeInBackDataBase(sql);

              log.info("load table=" + tabName + " successful");

6、数据表或文件下载的实现

由于存储在hdfs上的数据为Gzip压缩格式，首先通过执行事先编好的解压代码对文件块进行解压，这里需要传入需要解压的文件名、解压类型、hdfs的完全路径，解压完毕后通过执行hadoop文件合并命令将文件从hdfs上合并到本地服务器，合并完毕后由于解压缩后的文件会占用hdfs的空间，同时执行hadoop文件删除命令将解压后的文件删除，再通过ftp传到前台服务器，完成客户群清单下载。

String command = "cd " + ftpPath + ";" + hadoopPath + "hadoop jar "+hadoopPath+"CodecTable.jar " + listRandomName +" "+ CodecType

+" " + " "+ hdfsWholePath + ";" + hadoopPath + "hadoop fs -cat '" +  hdfsPath + listRandomName + "/*'>" + listName1+".csv;" + hadoopPath +"hadoop fs -rm -r " + hdfsPath + listRandomName + ";" + "wc -l " + listName1 +".csv;";

       LOG.debug(command);

 flag = ShellUtils.execCmd(command, user, passwd, host,num);

清单的推送也是通过文件合并传输的方式进行其他平台的推送，大大降低了读取数据插入表数据所消耗的时间。