数据清洗
...大约 4 分钟
数据清洗
一、背景
-- 1. 清洗的原因：
在数仓项目中，数据经常不是我们想要的，数据可能存在不完整的情况，或有一些null，或者格式不对，那么我们需要对数据进行清洗。
-- 2. 本案例的事件：
--用户数据：
'一条数据'
barelypolitical 151 5106
'每个字段代表的含义'
uploader:barelypolitical
videos: 151
friends: 5106
--viedo数据：
'一条数据'：
LKh7zAJ4nwo TheReceptionist 653 Entertainment 424 13021 4.34 1305 744 DjdA-5oKYFQ NxTDlnOuybo
c-8VuICzXtU DH56yrIO5nI
'每个字段代表的含义'：
'videoId'：LKh7zAJ4nwo
'uploader'：TheReceptionist
'age'：653
'category'：People & Blogs --(Array`<String>`)
'length'：424
'views'：13021
'rate'：4.34
'Ratings'：1305
'conments'：744
'relatedId'：DjdA-5oKYFQ NxTDlnOuybo c-8VuICzXtU DH56yrIO5nI -- 相关视频id(Array`<String>`)
-- 3. 我们需要做的清洗工作是：
1. 格式转换成统一的格式
视频类别的分隔符和相关视频的id两个数据格式进行处理。
2. 进行一些数据处理，数据格式不完整的不要。| 字段 | 备注 | 详细描述 |
|---|---|---|
| videoId | 视频唯一id(String) | 11位字符串 |
| uploader | 视频上传者(String) | 上传视频的用户名String |
| age | 视频年龄(int) | 视频在平台上的整数天 |
| category | 视频类别(Array<String>) | 上传视频指定的视频分类 |
| length | 视频长度(Int) | 整形数字标识的视频长度 |
| views | 观看次数(Int) | 视频被浏览的次数 |
| rate | 视频评分(Double) | 满分5分 |
| Ratings | 流量(Int) | 视频的流量，整型数字 |
| conments | 评论数(Int) | 一个视频的整数评论数 |
| relatedId | 相关视频id(Array<String>) | 相关视频的id，最多20个 |
| *字段* | *备注* | *字段类型* |
|---|---|---|
| uploader | 上传者用户名 | string |
| videos | 上传视频数 | int |
| friends | 朋友数量 | int |
二、数据清洗思想
-- 1. 使用MR对数据进行清洗，由于不需要聚合，那么只需要map阶段就可以了
-- 2. 清洗的规则：
a、 数据长度小于9的不要
b、 将数据中的视频类别中间的空格去掉 People & Blogs
c、 将数据中的关联视频id通过&符号拼接三、代码实现
3.0 准备工作
- 添加依赖
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<version>2.12.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>3.1.3</version>
</dependency>
</dependencies>- 在resource目录下创建一个log4j2.xml文件，并添加如下配置内容
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="error" strict="true" name="XMLConfig">
<Appenders>
<!-- 类型名为Console，名称为必须属性 -->
<Appender type="Console" name="STDOUT">
<!-- 布局为PatternLayout的方式，
输出样式为[INFO] [2018-01-22 17:34:01][org.test.Console]I'm here -->
<Layout type="PatternLayout"
pattern="[%p] [%d{yyyy-MM-dd HH:mm:ss}][%c{10}]%m%n" />
</Appender>
</Appenders>
<Loggers>
<!-- 可加性为false -->
<Logger name="test" level="info" additivity="false">
<AppenderRef ref="STDOUT" />
</Logger>
<!-- root loggerConfig设置 -->
<Root level="info">
<AppenderRef ref="STDOUT" />
</Root>
</Loggers>
</Configuration>3.1 创建一个工具类
/**
* @author lianzhipeng
* @Description
* @create 2020-07-20 2:21:30
*/
public class ETLUtils {
/**
* 数据清洗方法
*/
public static String etlData(String srcData){
StringBuffer resultData = new StringBuffer();
//1. 先将数据通过\t 切割
String[] datas = srcData.split("\t");
/**
* LKh7zAJ4nwo
* TheReceptionist
* 653
* People & Blogs
* 424
* 13021
* 4.34
* 1305
* 744
* DjdA-5oKYFQ NxTDlnOuybo c-8VuICzXtU DH56yrIO5nI
*/
//2. 判断长度是否小于9
if(datas.length <9){
return null ;
}
//3. 将数据中的视频类别的空格去掉
datas[3]=datas[3].replaceAll(" ","");
//4. 将数据中的关联视频id通过&拼接
for (int i = 0; i < datas.length; i++) {
if(i < 9){
//4.1 没有关联视频的情况
if(i == datas.length-1){
resultData.append(datas[i]);
}else{
resultData.append(datas[i]).append("\t");
}
}else{
//4.2 有关联视频的情况
if(i == datas.length-1){
resultData.append(datas[i]);
}else{
resultData.append(datas[i]).append("&");
}
}
}
return resultData.toString();
}
}3.2 创建mapper类
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
/**
* @author lianzhipeng
* @Description
* @create 2020-07-20 2:21:42
*/
public class ETLMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
private Text k = new Text();
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//获取一行
String line = value.toString();
//清洗
String resultData = ETLUtils.etlData(line);
if(resultData != null) {
//写出
k.set(resultData);
context.write(k,NullWritable.get());
}
}
}3.3 创建driver类
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
/**
* @author lianzhipeng
* @Description
* @create 2020-07-20 2:21:53
*/
public class ETLDriver {
public static void main(String[] args) throws Exception {
// 创建一个配置文件的对象
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
// 设定驱动类
job.setJarByClass(ETLDriver.class);
// 设置map类
job.setMapperClass(ETLMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(NullWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
job.setNumReduceTasks(0);
FileInputFormat.setInputPaths(job,new Path(args[0]));
FileOutputFormat.setOutputPath(job,new Path(args[1]));
job.waitForCompletion(true);
}
}四、代码打包
将ETL程序打包为etl.jar 并上传到Linux的 /opt/module/hive/datas 目录下
赞助
