ETL实践--Spark做数据清洗

 

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&amp;useUnicode=true&amp;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);

}

}

 

    原文作者:spark
    原文地址: https://www.cnblogs.com/double-kill/p/8283751.html
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞