ETL实践–Spark做数据清洗
上篇博客，说的是用hive代替kettle的表关联。是为了提高效率。
本文要说的spark就不光是为了效率的问题。
1、用spark的原因
（如果是一个sql能搞定的关联操作，可以直接用kettle导原始数据到hive，用hive视图做关联直接给kylin提供数据）
(1)、场景一之前用kettle需要多个转换、关联才能实现数据清洗的操作。
用hive不知道如何进行，就算能进行也感觉繁琐，同时多个步骤必然降低数据时效性。用mr的话也是同样道理太多步骤繁琐不堪。
(2)、一些不能用sql来处理的数据清洗逻辑。比如循环类的，或者是更复杂的处理逻辑。用hive和kettle都不方便解决。
一些其他的原因
(3)、支持的语言多，容易上手，并且之前也学习过一些。
(4)、我公司用的大数据平台上，提供了spark的支持，可以方便的安装和维护，并且可以和现有平台很好的融合（yarn部署方式）。
(5)、效率高。
(6)、刚好公司有需要用到spark streaming。
2、下面是我学习用spark处理业务问题的一个例子。有注释和一些方法的测试。
public class EtlSpark5sp2Demo {
//3、解码器
//static final Encoder<EcardAccessOutTime> outTimeEncoder = Encoders.bean(EcardAccessOutTime.class);
static final Encoder<EcardAccessInout> inoutEncoder = Encoders.bean(EcardAccessInout.class);
public static void main(String[] args) {
SparkSession spark = SparkSession
.builder().master("local")
.appName("Java Spark SQL data sources example")
.config("spark.some.config.option", "some-value")
.getOrCreate();
runJdbcDatasetExample(spark);
spark.stop();
}
private static void runJdbcDatasetExample(SparkSession spark) {
Properties connectionProperties = new Properties();
connectionProperties.put("user", "root123");
connectionProperties.put("password", "123");
String dbUrl="jdbc:mysql://192.168.100.4:3306/datasql?autoReconnect=true&useUnicode=true&characterEncoding=UTF-8";
//对比java中连接mysql的字符串（标红的是区别，分割字符串是spark识别不了的，要去掉）
// mysql.url=jdbc:mysql://192.168.100.4:3306/datasql?autoReconnect=true&useUnicode=true&characterEncoding=UTF-8
//1、查询原始数据
final Dataset<Row> allRecord = spark.read()
.jdbc(dbUrl,
"( select a.kid,a.outid,a.ioflag,a.OpDT as 'indate',a.OpDT as 'outdate',a.school_code ,a.faculty_code,a.major_code,a.class_code,a.sex from access_record_inout_temp2 a limit 3000 ) t", connectionProperties);
final Dataset<EcardAccessInout> allRecordInout=allRecord.as(inoutEncoder);
spark提供了把数据集注册成视图，然后用sql的方式对数据集进行处理的功能：如下2、3所示
//2、将原始数据注册成视图
allRecord.createOrReplaceTempView("view_access_record_inout_temp2");
allRecord.printSchema();//打印数据集结构
//3、在上面注册的视图上执行sql测试：查询出进入宿舍的记录
final Dataset<EcardAccessInout> inRecord = spark.sql(
" select a.kid,a.outid,a.ioflag,a.school_code,a.faculty_code,a.major_code,a.class_code,a.sex,a.indate,a.outdate " +
" from view_access_record_inout_temp2 a " +
" where a.ioflag = 0 ").as(inoutEncoder);
inRecord.printSchema();//打印数据集结构
inRecord.show();
spark不光提供了针对注册的视图的sql查询。也提供了通过方法来查询数据集的方式：下面是2种方式
//4、filter用法测试：
String outid="45723107";
long kid=7516452;
//filter方法能够正常使用
Dataset<Row> list1 = allRecord.filter(allRecord.col("outid").equalTo(outid).and(allRecord.col("kid").gt(kid))).orderBy(allRecord.col("indate"));//.take(1);//
list1.show();//打印前20条记录
5、where用法测试
//where方法能够正常使用
Dataset<Row> list2 = allRecord.where("outid = '"+ outid +"'").where("kid > "+ kid +"").orderBy("indate");
list2.show();//打印前20条记录
//6、分组取topN:测试：：mysql中可以group by 2个字段查询全部的字段。实际返回值是取的分组后的第一条记录。（对应实际业务就按照学号和时间去重，数据当中有重复数据）6.1、mysql中的原始sql
select a.id,a.outid,g.school_code,g.faculty_code,g.major_code,g.class_code,g.sex,a.OpDT,a.ioflag from access_record a inner join own_org_student g on a.OutId=g.outid
where a.id > ? group by a.OutId,a.OpDT
6.2、hive和spark都不支持这种操作。他们的做法是一样的。就是通过下面这个sql，用row_number()函数分组，取第一 final Dataset<Row> topRecord = spark.sql(
" select t.kid,t.outid,t.ioflag,t.school_code,t.faculty_code,t.major_code,t.class_code,t.sex,t.indate,t.outdate from (" +
" select a.kid,a.outid,a.ioflag,a.school_code,a.faculty_code,a.major_code,a.class_code,a.sex,a.indate,a.outdate, " +
" row_number() over(partition by outid order by indate) as rowNumber " + //根据行号top
" from view_access_record_inout_temp2 a " +
" where a.ioflag = 0 " +
" ) t where rowNumber =1 ");
topRecord.show();//打印前20条记录
//7、循环：合并进入记录和出去的记录
7.1、这里先进行先按照学号分区，再按照时间排序。
Dataset<EcardAccessInout> allRecordSort= allRecordInout.repartition(allRecordInout.col("outid")).sortWithinPartitions(allRecordInout.col("indate"));
7、循环：合并进入记录和出去的记录（因为已经排序了，本条记录的下一条，如果是正常记录就是出去的记录） Iterator<EcardAccessInout> iterator = allRecordSort.toLocalIterator();
List<EcardAccessInout> result=new ArrayList<EcardAccessInout>();
while(iterator.hasNext()){
EcardAccessInout first= iterator.next();
if("0".endsWith(first.getIoflag())){//第一条记录是：进入宿舍的记录
if(iterator.hasNext()){
EcardAccessInout second= iterator.next();
//取比入记录大的最小的一条出记录的时间，作为入记录的出时间。（排序后，后面一条就是最小记录）
if(first.getOutid().endsWith(second.getOutid())&&("1".endsWith(second.getIoflag()))&&first.getIndate().before(second.getIndate())){
first.setOutdate(second.getIndate());
result.add(first);
}
}
}
}
//8、处理后的数据写入mysql。这里只是个例子，实际数据应该是写到hdfs，变成hive表
inRecord.write().mode(SaveMode.Append)
.jdbc(dbUrl, "datacenter.access_record_inout_temp10", connectionProperties);
}
}
