使用Spark对数据进行ETL并存放到HBase中流程总结

傷城~ 2021-11-09 01:16 536阅读 0赞

# ETL #

ETL用来描述将数据从来源端经过萃取（extract）、转置（transform）、加载（load）至目的端的过程

# 使用场景 #

要对某些日志数据文件进行分析时  
ETL这一步是避免不了的  
因为采集到的日志数据，总会有一部分脏数据  
这部分脏数据可能会对后续的指标分析造成影响  
所以要对数据进行一些操作，将数据转换成比较有规律的  
我们想要的数据

# 使用Spark对数据进行ETL操作的一些步骤总结 #

## 第一步：读取采集到的日志文件 ##

采集日志文件使用的工具有很多，如flume、kafka、ELk等工具  
这些工具可以将采集到的日志数据存放到相应的文件系统中  
如Flume可以将采集到的数据存放到HDFS，HIVE中  
使用Spark可以从HDFS或者HBase中读取数据  
也可以从本地读取到需要处理的日志文件

## 第二步：解析读取到的日志数据 ##

使用Spark的SparkContext或SparkSession可以从不同位置读取到数据  
都会形成一个RDD（Resilient Distributed Datasets),弹性分布式数据集  
RDD可以说是Spark的核心数据集合  
此处常使用map方法操作读取到的RDD，将其中的数据转换成想要的数据格式

### 在这一步可以对数据进行解析 ###

如根据ip地址，分析出用户所属地域信息  
如果是浏览器端的项目，可以解析出操作系统的信息，浏览器的信息  
对时间格式进行转化  
对埋点数据进行解析，分析出埋点对应的类型或事件类型

## 第三步：Driver发送Executor需要的数据 ##

有些数据存放在Drive上，但是Executor在执行任务是需要这些数据  
为了更好的节省内存，可以使用Driver将该数据发送到Executor上  
Executor会创建一个独立的内存空间，存放该数据，作为只读的共享数据

## 第四步：过滤第二步得到的RDD数据集合 ##

在这一步，可以将对数据进行过滤  
将我们不需要的数据进行过滤  
使用RDD.filter方法进行过滤

## 第五步：处理第四步经过过滤的数据，返回需要的数据 ##

在这一步，通常进行的操作即是将我们需要的数据  
从中提取出来，因为数据可能会有几十项  
但需要的数据只有几项，此时就可以将需要的数据提取出来  
避免分析的数据太过庞大  
返回的大多是二元组或者多元组

## 第六步：将数据存放到HBase中 ##

第六步是在第五步中执行的  
为了更好的理清思路，将其分步写出  
这一步建立在第五步的基础之上  
在第五步截取出需要的数据后  
要考虑存放到HBase中时，rowkey的问题  
在这一步就要将rowkey设计并生成  
rowkey生成要满足三个点：唯一、热点、前缀匹配  
同时在这一步，要将生成的rowkey进行转换  
转换为ImmutableBytesWritable类型  
还用生成一个Put对象，将截取出的数据进行Bytes转换后放到Put对象中  
转换后的rowkey也要放到改put对象中  
这一步的操作是在map中进行的  
所以最后返回的值类型是一个二元组  
key就是rowkey，value就是Put对象

## 第七步：HBase表的建立，并将第六步中的数据存放到HBase表中 ##

建立HBase表时，要注意表名的设置，一般是每天都要分析数据  
所以建议表名为日期，可以避免表名重复  
或者将其作为参数，在执行Spark程序时将表面传入  
同时设置表的数据压缩方式，并建立预分区  
在建立HBase表时，会用HBaseConfiguration、HBaseAdmin这两个类  
一个用来获取HBase连接，一个用来执行建表的操作  
再向HBase中存放数据时，要设置一些输出的格式

// 设置输出OutputFormat 
        conf.set("mapreduce.job.outputformat.class","org.apache.hadoop.hbase.mapreduce.TableOutputFormat")
        // 设置保存到HBase表的名称
        conf.set(TableOutputFormat.OUTPUT_TABLE, tableName)
        // 设置输出目录
        conf.set("mapreduce.output.fileoutputformat.outputdir","/datas/spark/hbase/etl-output" + System.currentTimeMillis())

### 保存数据到表中 ###

是将第六步操作后的RDD数据存放到HBase中  
会用到RDD.saveAsNewAPIHadoopDataset（）方法将数据存放到HBase中

## 第八步 ##

关闭SparkContext或者SparkSession获取到的资源