热门标签 | HotTags
当前位置:  开发笔记 > 运维 > 正文

HadoopPigLoadfunc

hadooppig设计的还是很不错的,可以写UDF每一个统计基本上都是要对原始日志进行切分,把想要的一些字段EXTRACT提取出来日志有着基本的模式mac:50:A4:C8:D7:10:7D|build:5141bc99|network:mobile|version:2.4.1|id:taobao22935952431|基本

hadoop pig 设计的还是很不错的,可以写 UDF 每一个统计基本上都是要对原始日志进行切分,把想要的一些字段 EXTRACT 提取出来 日志有着基本的模式 "mac:50:A4:C8:D7:10:7D"|"build:5141bc99"|"network:mobile"|"version:2.4.1"|"id:taobao22935952431"| 基本

hadoop pig 设计的还是很不错的,可以写 UDF

每一个统计基本上都是要对原始日志进行切分,把想要的一些字段 EXTRACT 提取出来

日志有着基本的模式

"mac:50:A4:C8:D7:10:7D"|"build:5141bc99"|"network:mobile"|"version:2.4.1"|"id:taobao22935952431"|

基本上是 key, value对,自定义一个 load function ,指定 key,就可以获取 对应的value,在 pig 中可以使用

REGISTER /jar/kload.jar;
AA = LOAD '/log/load.log' USING kload.KoudaiLoader('mac,build') AS (mac,build);
DUMP AA;

输出结果

(50:A4:C8:D7:10:7D,5141bc99)

koudaiLoader是自己实现的一个 Load function,输出为要获取的key,输出为key所对应的 value

    package kload;
    import java.io.IOException;
    import java.util.*;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.*;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.pig.*;
    import org.apache.pig.backend.executionengine.ExecException;
    import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.*;
    import org.apache.pig.data.*;
    public class KoudaiLoader  extends LoadFunc{
            protected RecordReader recordReader = null;
            private String fieldDel = "";
            private String[] reqFildList;
            private ArrayList mProtoTuple = null;
            private TupleFactory mTupleFactory = TupleFactory.getInstance();
            private static final int BUFFER_SIZE = 1024;
            public KoudaiLoader() {
            }
            public KoudaiLoader(String delimiter) {
                    this();
                    if(delimiter == null || delimiter.length() == 0){
                            throw new RuntimeException("empty delimiter");
                    }
                    this.reqFildList=delimiter.split(",");
            }
            @Override
            public Tuple getNext() throws IOException {
                    try {
                            Map tmpMap = new HashMap();
                            List lst = new ArrayList();
                            boolean flag = recordReader.nextKeyValue();
                            int i = 0;
                            if (!flag) {
                                    return null;
                            }
                            Text value = (Text) recordReader.getCurrentValue();
                            tmpMap = this.sourceToMap(value.toString());
                            if( tmpMap == null || tmpMap.size() == 0 ){
                                return null;
                            }
                            for (String s :this.reqFildList){
                                String item = tmpMap.get(s); 
                                if(item == null || item.length() == 0){
                                        item = "";
                                }
                                    lst.add(i++, item);
                            }
                            return TupleFactory.getInstance().newTuple(lst);
                    } catch (InterruptedException e) {
                            throw new ExecException("Read data error", PigException.REMOTE_ENVIRONMENT, e);
                    }
            }
            public Map sourceToMap(String pline){
                    String line = pline;
                    int strLen = 0;
                    String[] strArr;
                    String[] strSubArr;
                    Map mapLog = new HashMap();
                    if(pline == null || pline.length() <= 2){
                        return mapLog;
                    }
                    line = line.trim();
                    strLen = line.length();
                    line = line.substring(1,strLen -1);
                    strArr = line.split("\"\\|\"");
                    if(strArr.length == 0){
                        return mapLog;
                    }
                    for(String s:strArr){
                            if(s != null && s.length() != 0){
                                    strSubArr = s.split(":",2); 
                                    if(strSubArr.length == 2){
                                            mapLog.put(strSubArr[0],strSubArr[1]);
                                    }
                            }
                    }
                    return mapLog;
            }
            @Override
            public void setLocation(String s, Job job) throws IOException {
                FileInputFormat.setInputPaths(job, s);
            }
            @Override
            public InputFormat getInputFormat() throws IOException {
                return new PigTextInputFormat();
            }
            @Override
            public void prepareToRead(RecordReader recordReader, PigSplit pigSplit) throws IOException {
                this.recordReader = recordReader;
            }
    }

编译

javac -cp /usr/local/webserver/pig/pig-0.9.2.jar:.  KoudaiLoader.java

打成jar包

jar -cf kload.jar kload

用pig在本地模式下运行

java -cp /usr/local/webserver/pig/pig-0.9.2.jar:/jar/kload.jar org.apache.pig.Main -x local kload.pig
推荐阅读
  • HBase运维工具全解析
    本文深入探讨了HBase常用的运维工具,详细介绍了每种工具的功能、使用场景及操作示例。对于HBase的开发人员和运维工程师来说,这些工具是日常管理和故障排查的重要手段。 ... [详细]
  • Hadoop发行版本选择指南:技术解析与应用实践
    本文详细介绍了Hadoop的不同发行版本及其特点,帮助读者根据实际需求选择最合适的Hadoop版本。内容涵盖Apache Hadoop、Cloudera CDH等主流版本的特性及应用场景。 ... [详细]
  • 深入解析Hadoop的核心组件与工作原理
    本文详细介绍了Hadoop的三大核心组件:分布式文件系统HDFS、资源管理器YARN和分布式计算框架MapReduce。通过分析这些组件的工作机制,帮助读者更好地理解Hadoop的架构及其在大数据处理中的应用。 ... [详细]
  • 深入解析Spark核心架构与部署策略
    本文详细探讨了Spark的核心架构,包括其运行机制、任务调度和内存管理等方面,以及四种主要的部署模式:Standalone、Apache Mesos、Hadoop YARN和Kubernetes。通过本文,读者可以深入了解Spark的工作原理及其在不同环境下的部署方式。 ... [详细]
  • Hadoop MapReduce 实战案例:手机流量使用统计分析
    本文通过一个具体的Hadoop MapReduce案例,详细介绍了如何利用MapReduce框架来统计和分析手机用户的流量使用情况,包括上行和下行流量的计算以及总流量的汇总。 ... [详细]
  • HBase 数据复制与灾备同步策略
    本文探讨了HBase在企业级应用中的数据复制与灾备同步解决方案,包括存量数据迁移及增量数据实时同步的方法。 ... [详细]
  • 流处理中的计数挑战与解决方案
    本文探讨了在流处理中进行计数的各种技术和挑战,并基于作者在2016年圣何塞举行的Hadoop World大会上的演讲进行了深入分析。文章不仅介绍了传统批处理和Lambda架构的局限性,还详细探讨了流处理架构的优势及其在现代大数据应用中的重要作用。 ... [详细]
  • 本文介绍了如何在 MapReduce 作业中使用 SequenceFileOutputFormat 生成 SequenceFile 文件,并详细解释了 SequenceFile 的结构和用途。 ... [详细]
  • 本文介绍了如何使用Flume从Linux文件系统收集日志并存储到HDFS,然后通过MapReduce清洗数据,使用Hive进行数据分析,并最终通过Sqoop将结果导出到MySQL数据库。 ... [详细]
  • 深入浅出:Hadoop架构详解
    Hadoop作为大数据处理的核心技术,包含了一系列组件如HDFS(分布式文件系统)、YARN(资源管理框架)和MapReduce(并行计算模型)。本文将通过实例解析Hadoop的工作原理及其优势。 ... [详细]
  • 本文介绍如何通过整合SparkSQL与Hive来构建高效的用户画像环境,提高数据处理速度和查询效率。 ... [详细]
  • 本文介绍了Hadoop的核心组件,包括高可靠性和高吞吐量的分布式文件系统HDFS、分布式的离线并行计算框架MapReduce、作业调度与集群资源管理框架YARN以及支持其他模块的工具模块Common。 ... [详细]
  • 大数据领域的职业路径与角色解析
    本文将深入探讨大数据领域的各种职业和工作角色,帮助读者全面了解大数据行业的需求、市场趋势,以及从入门到高级专业人士的职业发展路径。文章还将详细介绍不同公司对大数据人才的需求,并解析各岗位的具体职责、所需技能和经验。 ... [详细]
  • 从0到1搭建大数据平台
    从0到1搭建大数据平台 ... [详细]
  • Presto:高效即席查询引擎的深度解析与应用
    本文深入解析了Presto这一高效的即席查询引擎,详细探讨了其架构设计及其优缺点。Presto通过内存到内存的数据处理方式,显著提升了查询性能,相比传统的MapReduce查询,不仅减少了数据传输的延迟,还提高了查询的准确性和效率。然而,Presto在大规模数据处理和容错机制方面仍存在一定的局限性。本文还介绍了Presto在实际应用中的多种场景,展示了其在大数据分析领域的强大潜力。 ... [详细]
author-avatar
dsgfg
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有