ruby和pig处理流式文件实例

大数据操作中涉及到数据清洗步奏还是用脚本处理比较方便，下边介绍一下pig加载hdfs文件后调用ruby脚本处理数据，再返回数据流至pig中处理的一个简单案例。

注意：ruby的流式处理用到wukong这个gem包，相关下载：
https://github.com/mrflip/wukong

pig中加载分布式文件调用ruby流式处理：

代码如下:

log = load '$INFILE' using PigStorage('\t');

define tracking_parser `/usr/ruby parse_click.rb --map` SHIP('parse_click.rb', 'click_tracking.rb');

strmo = stream log through tra_parser;

store strmo into '$OUTFILE' using PigStorage('\t');

代码如下:

require 'wukong'
require 'json'
require './click_tra.rb'

module ParseClick
  class Mapper < Wukong::Streamer::RecordStreamer
    def before_stream
      @bad_count = 0
    end

def after_stream
      raise RuntimeError, "Exceeded bad records : #{@bad_count}" if @bad_count > 10
    end

def process *records
      yield ClickTra.new(JSON.parse(records[2])).to_a
    rescue => e
      @bad_count += 1
      warn "Bad record #{e}: #{records[2]}"
    end
  end
end

Wukong.run ParseClick::Mapper, nil

代码如下:

require 'date'
require './models.rb'

class ClickTra
 
  output :ip
  output :c_date
  #output your other atrributes

def c_date
    click_date.strftime("%Y%m%d").to_i
  end

def ip
    browser_ip.to_i
  end

end

其中

strmo = stream log through tra_parser;调用定义的外部程序tra_parser处理log对象。
Wukong.run ParseClick::Mapper, nil执行完后，将ruby执行结果回调pig接收。
store strmo into '$OUTFILE' using PigStorage('\t');做结果存储持久化。

赞 (0)

相关推荐

  • ruby和pig处理流式文件实例

    大数据操作中涉及到数据清洗步奏还是用脚本处理比较方便,下边介绍一下pig加载hdfs文件后调用ruby脚本处理数据,再返回数据流至pig中处理的一个简单案例. 注意:ruby的流式处理用到wukong这个gem包,相关下载: https://github.com/mrflip/wukong pig中加载分布式文件调用ruby流式处理: 复制代码 代码如下: log = load '$INFILE' using PigStorage('\t'); define tracking_parser `/

  • javascript瀑布流式图片懒加载实例解析与优化

    之前写过一版图片"懒加载"的文章,刚好周末在整理文件的时候,大概又看了一遍之前写的代码发现有很多可以优化的地方. 这篇文章主要就是结合上篇<javascript瀑布流式图片懒加载实例>再来看看图片"懒加载"的一些知识. 图片"懒加载"的主旨: 按照需要加载图片,也就是说需要显示的时候再加载图片显示,减少一次性加载的网络带宽开销. 先来看一段代码: var conf = { 'loadfirst': true, 'loadimg': t

  • Android自定义流式布局/自动换行布局实例

    最近,Google开源了一个流式排版库"FlexboxLayout",功能强大,支持多种排版方式,如各种方向的自动换行等,具体资料各位可搜索学习^_^. 由于我的项目中,只需要从左到右S型的自动换行,需求效果图如下: 使用FlexboxLayout这个框架未免显得有些臃肿,所以自己动手写了一个流式ViewGroup. 安卓中自定义ViewGroup的步骤是: 1. 新建一个类,继承ViewGroup 2. 重写构造方法 3. 重写onMeasure.onLayout方法 onMeasu

  • RestTemplate文件上传下载与大文件流式下载

    目录 一.文件上传 二.文件下载 三.大文件下载 本文是精讲RestTemplate第6篇,前篇的blog访问地址如下: RestTemplate是HTTP客户端库,所以为了使用RestTemplate进行文件上传和下载,需要我们先编写服务端的支持文件上传和下载的程序.请参考我之前写的一篇文章:SpringBoot实现本地存储文件上传及提供HTTP访问服务 .按照此文完成学习之后,可以获得 一个以访问服务URI为"/upload”的文件上传服务端点 服务端点上传文件成功后会返回一个HTTP连接,

  • golang gin框架实现大文件的流式上传功能

    目录 upload.html gin_stream_upload_file.go 一般来说,通过c.Request.FormFile()获取文件的时候,所有内容都全部读到了内存.如果是个巨大的文件,则可能内存会爆掉:且,有的时候我们需要一边上传一边处理.以下的代码实现了大文件流式上传.还非常不完美,但是可以作为参考: upload.html <!DOCTYPE html> <html lang="en"> <head> <meta charse

  • java使用RandomAccessFile类基于指针读写文件实例代码

    java API中提供了一个基于指针操作实现对文件随机访问操作的类,该类就是RandomAccessFile类,该类不同于其他很多基于流方式读写文件的类.它直接继承自Object. public class RandomAccessFile extends Objectimplements DataOutput, DataInput, Closeable{...} 1.使用该类时可以指定对要操作文件的读写模式. 第一种模式是只读模式,第二种模式是读写模式.在创建该类实例时指定. @Test pu

  • Android简单实现自定义流式布局的方法

    Android简单实现自定义流式布局的方法

    本文实例讲述了Android简单实现自定义流式布局的方法.分享给大家供大家参考,具体如下: 首先来看一下 手淘HD - 商品详情 - 选择商品属性 页面的UI 商品有很多尺码,而且展现每个尺码所需要的View的大小也不同(主要是宽度),所以在从服务器端拉到数据之前,展现所有尺码所需要的行数和每一行的个数都无法确定,因此不能直接使用GridView或ListView. 如果使用LinearLayout呢? 一个LinearLayout只能显示一行,如果要展示多行,则每一行都要new一个Linear

  • Android实现热门标签的流式布局

    Android实现热门标签的流式布局

    一.概述: 在日常的app使用中,我们会在android 的app中看见 热门标签等自动换行的流式布局,今天,我们就来看看如何 自定义一个类似热门标签那样的流式布局吧(源码下载在下面最后给出) 类似的自定义布局.下面我们就来详细介绍流式布局的应用特点以及用的的技术点: 1.流式布局的特点以及应用场景     特点:当上面一行的空间不够容纳新的TextView时候,     才开辟下一行的空间 原理图:    场景:主要用于关键词搜索或者热门标签等场景 2.自定义ViewGroup,重点重写下面两

  • Java Swing组件布局管理器之FlowLayout（流式布局）入门教程

    Java Swing组件布局管理器之FlowLayout（流式布局）入门教程

    本文实例讲述了Java Swing组件布局管理器之FlowLayout(流式布局).分享给大家供大家参考,具体如下: FlowLayout应该是Swing布局管理器学习中最简单.最基础的一个.所谓流式,就是内部控件像水流一样,从前到后按顺序水平排列,直到达到容器的宽度时跳转到第二行.既然是水平排列,那么就存在三种基本的对齐方式:居中对齐(CENTER ).左对齐(LEFT )和右对齐(RIGHT ).然而,FlowLayout还提供两种对齐方式:LEADING,表示控件与容器方向开始边对应:TR

  • Sanic框架流式传输操作示例

    本文实例讲述了Sanic框架流式传输操作.分享给大家供大家参考,具体如下: 简介 Sanic是一个类似Flask的Python 3.5+ Web服务器,它的写入速度非常快.除了Flask之外,Sanic还支持异步请求处理程序.这意味着你可以使用Python 3.5中新的闪亮的异步/等待语法,使你的代码非阻塞和快速. 在前面一篇<Sanic框架Cookies操作>中已经讲到,如何在Sanic中使用Cookie,接下来将介绍一下Sanic的流的使用: 请求流式传输 Sanic允许通过流获取请求数据

随机推荐