ETL数据清洗,最大亮点之自定义UDF/UDTF函数

左手的ㄟ右手 2022-10-10 04:51 98阅读 0赞

要对获取到的日志数据进行处理,去空,转义,清洗等等.这时候我就需要我们来对初始的原始数据进行清洗.日志服务器获取到的数据是往往json对象,因此要对json对象进行解析,通过自定义UDF/UDTF函数来对其进行清洗:  
(1)新建一个maven工程,在新建的maven工程中,创建好包.(com.atguigu.udf),然后再对pom.xml中导入如下的所需依赖:

<repositories>
            <repository>
                <id>spring</id>
                <url>https://maven.aliyun.com/repository/spring</url>
            </repository>
        </repositories>
        <properties>
            <project.build.sourceEncoding>UTF8</project.build.sourceEncoding>
            <hive.version>1.2.1</hive.version>
        </properties>
        <dependencies>
            <dependency>
                <groupId>org.apache.hive</groupId>
                <artifactId>hive-exec</artifactId>
                <version>${hive.version}</version>
            </dependency>
        </dependencies>
        <build>
            <plugins>
                <plugin>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>2.3.2</version>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                    </configuration>
                </plugin>
                <plugin>
                    <artifactId>maven-assembly-plugin</artifactId>
                    <configuration>
                        <descriptorRefs>
                            <descriptorRef>jar-with-dependencies</descriptorRef>
                        </descriptorRefs>
                    </configuration>
                    <executions>
                        <execution>
                            <id>make-assembly</id>
                            <phase>package</phase>
                            <goals>
                                <goal>single</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </build>

UDF函数实现的思路:UDF函数有两个参数,一个是叫line也就是一条json格式的日志数据,另一个参数是表字段组成的keys集合.首先将keys集合进行切割成String数组.接着将line切割成包含时间戳和json格式的数据,判断切割好的数组长度是否为2和json格式的数据是否为空,如果不为空.则创建jsonObject对象开始进行处理.通过遍历keys的中元素,取出jsonObject对象中所对应的值,并将其添加到StringBuilder对象中,最后将StringBuilder对象转成String并且返回.

package com.atguigu.udf;
    
    import org.apache.commons.lang.StringUtils;
    import org.apache.hadoop.hive.ql.exec.UDF;
    import org.json.JSONException;
    import org.json.JSONObject;
    
    public class BaseFieldUDF extends UDF { 
        //line:整条的原始数据  jsonkeysString:公众字段keys的集合
        public String evaluate(String line,String jsonkeysString){ 
            //new 一个sb
            StringBuilder sb = new StringBuilder();
            //1.切割jsonkeys mid.uid,vc....
            String[] jsonkeys = jsonkeysString.split(",");
            //2.处理line 服务器时间|json
            String[] logContents = line.split("\\|");
            //3.合法性校验(判断空可以通过StringUtils.isBlank方法)
            if(logContents.length != 2 || StringUtils.isBlank(logContents[1])){ 
                return "";
            }
            //4.开始处理json
            try { 
                JSONObject jsonObject = new JSONObject(logContents[1]);
                if(jsonObject==null){ 
                    return "";
                }else{ 
                    JSONObject base = jsonObject.getJSONObject("cm");
                    if(base == null ){ 
                        return "";
                    }else{ 
                        //循环遍历取值
                        for(int i =0;i<jsonkeys.length;i++){ 
                            String fieldName = jsonkeys[i].trim();
                            if(base.has(fieldName)){ 
                                sb.append(base.getString(fieldName)).append("\t");
                            }else{ 
                                sb.append(" ").append("\t");
                            }
                        }
                        sb.append(jsonObject.getString("et")).append("\t");
                        sb.append(logContents[0]).append("\t");
                    }
                }
            } catch (JSONException e) { 
                e.printStackTrace();
            }
         return sb.toString();
        }
        //主方法主要是用来做测试
        public static void main(String[] args) { 
    
            String line = "1541217850324|{ \"cm\":{ \"mid\":\"m7856\",\"uid\":\"u8739\",\"ln\":\"-74.8\",\"sv\":\"V2.2.2\",\"os\":\"8.1.3\",\"g\":\"P7XC9126@gmail.com\",\"nw\":\"3G\",\"l\":\"es\",\"vc\":\"6\",\"hw\":\"640*960\",\"ar\":\"MX\",\"t\":\"1541204134250\",\"la\":\"-31.7\",\"md\":\"huawei-17\",\"vn\":\"1.1.2\",\"sr\":\"O\",\"ba\":\"Huawei\"},\"ap\":\"weather\",\"et\":[{ \"ett\":\"1541146624055\",\"en\":\"display\",\"kv\":{ \"newsid\":\"n4195\",\"copyright\":\"ESPN\",\"content_provider\":\"CNN\",\"extend2\":\"5\",\"action\":\"2\",\"extend1\":\"2\",\"place\":\"3\",\"showtype\":\"2\",\"category\":\"72\",\"newstype\":\"5\"}},{ \"ett\":\"1541213331817\",\"en\":\"loading\",\"kv\":{ \"extend2\":\"\",\"loading_time\":\"15\",\"action\":\"3\",\"extend1\":\"\",\"type1\":\"\",\"type\":\"3\",\"loading_way\":\"1\"}},{ \"ett\":\"1541126195645\",\"en\":\"ad\",\"kv\":{ \"entry\":\"3\",\"show_style\":\"0\",\"action\":\"2\",\"detail\":\"325\",\"source\":\"4\",\"behavior\":\"2\",\"content\":\"1\",\"newstype\":\"5\"}},{ \"ett\":\"1541202678812\",\"en\":\"notification\",\"kv\":{ \"ap_time\":\"1541184614380\",\"action\":\"3\",\"type\":\"4\",\"content\":\"\"}},{ \"ett\":\"1541194686688\",\"en\":\"active_background\",\"kv\":{ \"active_source\":\"3\"}}]}";
            String x = new BaseFieldUDF().evaluate(line, "mid,uid,vc,vn,l,sr,os,ar,md,ba,sv,g,hw,nw,ln,la,t");
            System.out.println(x);
        }
    
    }