本文通过上一节Hadoop离线项目之数据清洗开发的数据清洗jar包，对日志文件进行清洗，并把清洗后的结果移动到hive表里，并刷新元数据信息。并把这个过程写到shell脚本里。
 jar包路径：
 
/home/hadoop/app/hadoop-2.6.0-cdh5.7.0/lib/g6-hadoop-1.0.jar
 
jar包主程序：
 
com.ruozedata.hadoop.mapreduce.driver.LogETLDriver
 
hdfs上日志文件路径：
 
/g6/hadoop/accesslog/
 
hdfs上日志文件清洗后要放的路径：
 
/g6/hadoop/access/output/
 
hive外部表g6_access指定的路径：
 
/g6/hadoop/access/clear/
 
shell脚本代码：
 
#!/bin/bash
if [ $# != 1 ] ; then
echo "USAGE: g6-train-hadoop.sh <dateString>"
echo " e.g.: g6-train-hadoop.sh 20180717"
exit 1;
fi

process_date=$1

echo "--------------------step1: mapreduce etl--------------------"
hadoop jar /home/hadoop/app/hadoop-2.6.0-cdh5.7.0/lib/g6-hadoop-1.0.jar com.ruozedata.hadoop.mapreduce.driver.LogETLDriver /g6/hadoop/accesslog/${process_date}.log /g6/hadoop/access/output/day=${process_date}

echo "--------------------step2:mv data to DW--------------------"
hdfs dfs -rm -r /g6/hadoop/access/clear/day=${process_date}
hdfs dfs -mv  /g6/hadoop/access/output/day=${process_date} /g6/hadoop/access/clear/

echo "--------------------step4:flush meatadata--------------------"
hive -e "use g6_hadoop; alter table g6_access add if not exists partition(day='${process_date}');"
 
然后运行shell脚本：
 
[hadoop@10-9-140-90 shell]$ ./g6-train-hadoop.sh 
USAGE: g6-train-hadoop.sh <dateString>
 e.g.: g6-train-hadoop.sh 20180717
[hadoop@10-9-140-90 shell]$ ./g6-train-hadoop.sh 20180717
......此处省略
 
运行成功后，去hive里查看一下：
 
hive (g6_hadoop)> select * from g6_access where day=20180717 limit 1;
OK
cdn     region  level   time    ip      domain  url     traffic day
baidu   CN      E       20180717042142  156.89.48.178   v2.go2yd.com    http://v1.go2yd.com/user_upload/1531633977627104fdecdc68fe7a2c4b96b2226fd3f4c.mp4_bd.mp4       62109   20180717
Time taken: 0.42 seconds, Fetched: 1 row(s)
hive (g6_hadoop)> 

1、新建job
 2、插入shell脚本
 
	需要设置环境变量，E:\Anaconda3;E:\Anaconda3\Scripts;E:\Anaconda3\Library\bin
	需要添加.py文件默认打开方式为python.exe
 

 3、新建转换
 
	首先设置变量以便于动态读取时间戳和动态读取excel文件目录；
	在执行sql脚本中需要勾选执行每一行，变量替换，绑定参数？；
	在excel输入中选中的文件的文件/目录下使用${变量}，即可取到excel文件名。
 

 4、在数据库中建表，执行所有步骤即可。
 
	create TABLESPACE ts_flow_data LOGGING datafile 'E:\ORACLE\T_FLOW_DATA_01.DBF' SIZE 128M 
	AUTOEXTEND ON NEXT 128M MAXSIZE 20G;
	COMMIT;

	CREATE TABLE t_flow_data ( fid NUMBER, fcleandate TIMESTAMP(6), fbizdate TIMESTAMP(6), 
	fcoustomername  NVARCHAR2(255), fmaterialname NVARCHAR2(255), fmodel NVARCHAR2(255), 
	fqty NUMBER(28, 16), fprice NUMBER(19, 4), famount NUMBER(19, 4), fshengchanqiye NVARCHAR2(255), 
	flot NVARCHAR2(80), fmfg TIMESTAMP(6), fexp TIMESTAMP(6), fsupplier NVARCHAR2(255))
	TABLESPACE ts_flow_data;
	COMMIT;

	CREATE SEQUENCE seq_flow_data MINVALUE 1 NOMAXVALUE START WITH 1 
	INCREMENT BY 1 CACHE 100 ORDER;
	COMMIT;

	CREATE TRIGGER trg_flow_data
	BEFORE INSERT ON t_flow_data
	FOR EACH ROW
		BEGIN
		    SELECT
		        seq_flow_data.NEXTVAL
		    INTO :new.fid
		    FROM
		        dual;
		END;
	COMMIT;

数据
 
此次数据是用来导入HIVE表中，但是由于数据不规范，需要做清洗，
 下面的数据不仅有空出一行，并且每个需要录入的字段都有空格。
 
| 1 | 2 | 2016-06-03 |

| 1 | 3 | 2016-06-08 |

| 2 | 3 | 2016-06-08 |

| 3 | 4 | 2016-06-09 |
 
清洗方法
 
此次咱用shell的sed方法来做清洗，代码如下：
 
#!/bin/sh
echo 'input clearing file name?'
read name

#删除所有空格
sed -i s/[[:space:]]//g $name
#首尾替换
sed -i 's/^[|]*//g' $name
sed -i 's/[|]*$//g' $name
#去空行：
sed -i '/^\s*$/d' $name
#中间的换成逗号
sed -i 's/|/,/g' $name

#coding=utf-8
import numpy as np
import pandas as pd
na_list=['NO CLUE','N/A','0']#na包含的类型
data=pd.read_csv('311-service-requests.csv',na_values=na_list,dtype={'Incident Zip':str})#列值数据类型设置为字符串
#print(data['Incident Zip'].head())#--查看前几行，不改变源数据
#data['Incident Zip'].unique()#--查看唯一值，不改变源数据
#data['Incident Zip'][data['Incident Zip'].str.len()>5]#查看元素长度大于5的元素，不改变源数据
data['Incident Zip']=data['Incident Zip'].str.slice(0,5)#源数据重新赋值，改变源数据
#print(data['Incident Zip'])
data.loc[data['Incident Zip']=='00000','Incident Zip']=np.nan#值为0的元素设为nan，loc改变源数据
unique_zip=data['Incident Zip'].fillna(False).unique().astype('str')#fillna(Flase)
unique_zip.sort()#排序，改变源数据
#print(unique_zip)
zips=data['Incident Zip']
is_close=zips.str.startswith('0')|zips.str.startswith('1')#取字符传以1或0开头的元素
is_far=~(is_close)&zips.notna()#～表示反条件，即不以0或1开头且不为空的元素
print(data[is_far]['Incident Zip'])#花式索引