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处理业务问题的一个例子。有注释和一些方法的测试。

//3、解码器 //static final Encoder outTimeEncoder = Encoders.bean(EcardAccessOutTime.class); static final Encoder 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识别不了的，要去掉）

final Dataset allRecordInout=allRecord.as(inoutEncoder);

//4、filter用法测试：

//where方法能够正常使用

//6、分组取topN:测试：：mysql中可以group by 2个字段查询全部的字段。实际返回值是取的分组后的第一条记录。

                            from access_record a inner join own_org_student g on a.OutId=g.outid                                where a.id > ?   group by a.OutId,a.OpDT

topRecord.show();//打印前20条记录

//7、循环：合并进入记录和出去的记录 7.1、这里先进行先按照学号分区，再按照时间排序。 Dataset allRecordSort= allRecordInout.repartition(allRecordInout.col("outid")).sortWithinPartitions(allRecordInout.col("indate"));

//8、处理后的数据写入mysql。这里只是个例子，实际数据应该是写到hdfs，变成hive表 inRecord.write().mode(SaveMode.Append) .jdbc(dbUrl, "datacenter.access_record_inout_temp10", connectionProperties);

}

}