ETL实践--Spark作数据清洗java
上篇博客，说的是用hive代替kettle的表关联。是为了提升效率。mysql
本文要说的spark就不光是为了效率的问题。sql
一、用spark的缘由app
（若是是一个sql能搞定的关联操做，能够直接用kettle导原始数据到hive，用hive视图作关联直接给kylin提供数据）函数
(1)、场景一以前用kettle须要多个转换、关联才能实现数据清洗的操做。学习
用hive不知道如何进行，就算能进行也感受繁琐，同时多个步骤必然下降数据时效性。用mr的话也是一样道理太多步骤繁琐不堪。测试
(2)、一些不能用sql来处理的数据清洗逻辑。好比循环类的，或者是更复杂的处理逻辑。用hive和kettle都不方便解决。大数据
一些其余的缘由ui
(3)、支持的语言多，容易上手，而且以前也学习过一些。url
(4)、我公司用的大数据平台上，提供了spark的支持，能够方便的安装和维护，而且能够和现有平台很好的融合（yarn部署方式）。
(5)、效率高。
(6)、恰好公司有须要用到spark streaming。
二、下面是我学习用spark处理业务问题的一个例子。有注释和一些方法的测试。
public class EtlSpark5sp2Demo {
//三、解码器
//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
//一、查询原始数据
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的方式对数据集进行处理的功能：以下二、3所示
//二、将原始数据注册成视图
allRecord.createOrReplaceTempView("view_access_record_inout_temp2");
allRecord.printSchema();//打印数据集结构
//三、在上面注册的视图上执行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种方式
//四、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条记录
五、where用法测试
//where方法可以正常使用
Dataset<Row> list2 = allRecord.where("outid = '"+ outid +"'").where("kid > "+ kid +"").orderBy("indate");
list2.show();//打印前20条记录
//六、分组取topN:测试：：mysql中能够group by 2个字段查询所有的字段。实际返回值是取的分组后的第一条记录。（对应实际业务就按照学号和时间去重，数据当中有重复数据）6.一、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.二、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.一、这里先进行先按照学号分区，再按照时间排序。
Dataset<EcardAccessInout> allRecordSort= allRecordInout.repartition(allRecordInout.col("outid")).sortWithinPartitions(allRecordInout.col("indate"));
七、循环：合并进入记录和出去的记录（由于已经排序了，本条记录的下一条，若是是正常记录就是出去的记录）
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); } } } } //八、处理后的数据写入mysql。这里只是个例子，实际数据应该是写到hdfs，变成hive表 inRecord.write().mode(SaveMode.Append) .jdbc(dbUrl, "datacenter.access_record_inout_temp10", connectionProperties); }}