本项目为示例程序，还有很多需要完善的地方。
转换、清洗开发主要分为 3 部分：
- 定义数据源，即后续操作的基础数据源；
- 定义清洗逻辑，即基于基础数据的处理，得到我们需要的结果；
- 定义数据持久化，即我们最终的结果数据存储持久化。
<?xml version="1.0" encoding="UTF-8"?> <logic> <!-- 替换参数定义，可用于后面开发进行简单参数化定义 --> <parameters> <parameter name="path" value="/Users/mars/Desktop/Data/tools" /> <parameter name="path1" value="file:///Users/mars/Desktop/Data/bank/bank/people" /> <parameter name="path2" value="file:///Users/mars/Desktop/Data/bank/bank/people_append" /> <parameter name="path3" value="file:///Users/mars/Desktop/Data/bank/bank/people_out1" /> <parameter name="path4" value="file:///Users/mars/Desktop/Data/bank/bank/people_out2" /> <parameter name="path5" value="file:///Users/mars/Desktop/Data/bank/bank/people_out3" /> </parameters> <base> <!-- ${path}: 上述paramters定义的替换参数 --> <logPath>${path}/logs</logPath> <metaPath>${path}/metapath</metaPath> <tempPath>${path}/tmp</tempPath> <schemaPath>${path}/schemapath</schemaPath> <database>test</database> </base> <!-- 用于后续转换、清洗等操作的基础数据 --> <sources> <source tableName="people1" type="parquet" partitions="1" cache="true" dataPath="${path1}"/> <source tableName="people2" type="parquet" partitions="1" cache="true" dataPath="${path2}"/> </sources> <!-- 基于 sources 中定义的基础数据进行的转换、清洗操作； 支持向上依赖 --> <transforms> <transform tableName="people_t" cache="true" save="true" type="sql"> <![CDATA[ select job, day, age, education from people1 union all select job, day, age, education from people2 ]]> </transform> <transform tableName="people_t2" cache="true" save="true" type="sql"> <![CDATA[ select job, day, age, education from people1 union all select job, day, age, education from people2 union all select job, day, age, education from people_t ]]> </transform> <transform tableName="people_t3" cache="true" save="true" type="sql"> <![CDATA[ select job, day, age, education from people1 union all select job, day, age, education from people_t2 ]]> </transform> </transforms> <!-- 基于基础数据转换、清洗完成后，需要最终落地的数据 --> <sinks> <sink tableName="people_t" type="parquet" saveMode="overwrite" savePath="${path3}"/> <sink tableName="people_t2" type="txt" partitions="1" delimiter=" || " saveMode="overwrite" savePath="${path4}"/> <sink tableName="people_t3" type="txt" coalesce="2" delimiter=" || " saveMode="overwrite" savePath="${path5}"/> </sinks> </logic>
部分参数说明：
tableName: 数据源读取后在 Spark-SQL 中的注册表名
type: 数据源的类别，csv、txt、parquet等
partitions: 数据读取后重新分区的数量
cache: 是否缓存表
dataPath: 数据存放路径，默认使用 HDFS 路径
schemaPath: 当数据源为 txt 类的文本文件时，需要指定数据的 schema，比如：列分隔符、列名、列值类型等定义
<!-- 用于后续转换、清洗等操作的基础数据 --> <sources> <source tableName="people1" type="csv" partitions="1" cache="true" dataPath="${path1}" schemaPath="/path/to/people.meta"/> <source tableName="people2" type="parquet" partitions="1" cache="true" dataPath="${path2}"/> </sources>
people.meta
文本类数据源的结构映射定义，主要定义有数据的编码格式、列分隔符、每列的列名、列值类型等
<?xml version="1.0" encoding="UTF-8"?> <source charset="UTF-8" delimiter="\|\|" isHeaderFirst="true"> <columns> <column type="STRING" nullable="true" index="1"> age </column> <column type="STRING" nullable="true" index="2"> job </column> <column type="STRING" nullable="true" index="3"> marital </column> <column type="STRING" nullable="true" index="4"> education </column> <column type="STRING" nullable="true" index="5"> default </column> <column type="STRING" nullable="true" index="6"> balance </column> <column type="STRING" nullable="true" index="7"> housing </column> <column type="STRING" nullable="true" index="8"> loan </column> <column type="STRING" nullable="true" index="9"> contact </column> <column type="STRING" nullable="true" index="10"> day </column> <column type="STRING" nullable="true" index="11"> month </column> <column type="STRING" nullable="true" index="12"> duration </column> <column type="STRING" nullable="true" index="13"> campaign </column> <column type="STRING" nullable="true" index="14"> pdays </column> <column type="STRING" nullable="true" index="15"> previous </column> <column type="STRING" nullable="true" index="16"> poutcome </column> <column type="STRING" nullable="true" index="17"> y </column> </columns> </source>
基于上面来源数据表定义，进行我们期望的数据转换、清洗操作。
部分参数说明：
tableName: SQL执行完成后的数据注册成的表名
cache: 是否缓存表
save: 是否对此中间表存储
type: 当前只支持sql
<!-- 基于 sources 中定义的基础数据进行的转换、清洗操作； 支持向上依赖 --> <transforms> <transform tableName="people_t" cache="true" save="true" type="sql"> <![CDATA[ select job, day, age, education from people1 union all select job, day, age, education from people2 ]]> </transform> <transform tableName="people_t2" cache="true" save="true" type="sql"> <![CDATA[ select job, day, age, education from people1 union all select job, day, age, education from people2 union all select job, day, age, education from people_t ]]> </transform> <transform tableName="people_t3" cache="true" save="true" type="sql"> <![CDATA[ select job, day, age, education from people1 union all select job, day, age, education from people_t2 ]]> </transform> </transforms>
对于基于 sources 层和 transforms 层得到的表中最终需要持久化的表进行持久化操作，也就是整个逻辑的结果持久化。
部分参数说明：
tableName: sources和transforms中需要持久化的表
type: 持久化时数据类型，csv、txt、parquet 等
saveMode: 持久化动作，append、overwrite 等
partitions: 分区数，即生成的文件数
partitionBy: 分区字段
<!-- 基于基础数据转换、清洗完成后，需要最终落地的数据 --> <sinks> <sink tableName="people_t" type="parquet" saveMode="overwrite" savePath="${path3}"/> <sink tableName="people_t2" type="txt" partitions="1" delimiter=" || " saveMode="overwrite" savePath="${path4}"/> <sink tableName="people_t3" type="txt" coalesce="2" delimiter=" || " saveMode="overwrite" savePath="${path5}"/> </sinks>
通过 spark-submit 执行进行提交即可
spark-submit \ --jars ... \ --class com.moon.Run frog.jar \ --transform-file ./transform.xml