storm的数据源编程单元Spout学习整理

简介: Spout呢,是Topology中数据流的源头,也是Storm针对数据源的编程单元。一般数据的来源,是通过外部数据源来读取数据项(Tuple),并读取的数据项传输至作业的其他组件。编程人员一般可通过OutputFieldsDeclarer类的declareStream()方法来声明多个流,指定数据将要发送的流,然后使用SpoutOutputCollector的emit方法将数据发送。

Spout呢,是Topology中数据流的源头,也是Storm针对数据源的编程单元。一般数据的来源,是通过外部数据源来读取数据项(Tuple),并读取的数据项传输至作业的其他组件。编程人员一般可通过OutputFieldsDeclarer类的declareStream()方法来声明多个流,指定数据将要发送的流,然后使用SpoutOutputCollector的emit方法将数据发送。

这里整理了下ISpout和IComponent接口。

ISpout声明了Spout的核心方法,用于向Topology供给数据项。对于每一个发出的数据项,Storm通过Spout,可以追踪它经历处理过程的有向无环图(竟然也是DAG)。

void open (java,util.Map conf,TopologyContext context,SpoutOutputCollector collector)
用于实例化Spout的一个运行时任务,被急群众的某一进程调用 (conf对象维护Storm中针对该Spout的配置信息,context是一个上下文对象,可用于获取该组件运行时任务的信息,collector用于从该Spout发送数据项)

void close() 
用于停止一个Spout

void activate()
在Spout从非激活状态转换为激活状态时被调用

void deactivate()
在Spout的非激活状态被调用

void ack(java.lang.Object msgId)
Storm用于确认该Spout发送的这个数据项已经被完整处理
void fail(java.lang.Object msgId) Storm用于确认该Spout发送的这个数据项已经失败 void nextTuple() 当这个方法被调用时,Storm要求Spout发送一个数据项至output collector

(nextTuple是Spout向Topology中发送一个数据项,是Spout需要实现的最重要的方法。在可靠的Spout的一个任务中,nextTuple()、ack()、fail()三个方法的调用在一个单独线程中循环。当不存在数据项需要发送时,nextTuple()将会休眠一小段间隔,确保不会浪费过多的CPU资源)

 

IComponent接口,声明了Topology组件的通用方法。使用JAVA语言的Spout和Bolt都必须实现这个接口。

void declareOutputFields(OutputFieldsDeclarer declarer)
声明指定输出流的数据项结构。(这里指定了输出流的数据项结构(schema)。参数declarer被用来声明输出流(stream)的id,域。

java.util.Map  getComponentConfiguration()
获取组件的配置信息

 

以Storm官网的WordCount来说明就是:

public class WordCount extends BaseRichSpout
{
    public static Logger log = logger.getLogger(backtype/storm/testing/WordCount);

    boolean_isDistributed;
    SpoutOutputCollector_collector;
    
    public WordCount(){
        this(true);
    }
    
    public WordCount(boolean isDistributed){
        _isDistributed = isDistributed;
    }

    public void open(Map conf,TopologyContext context,SpoutOutputCollector collector){
    _collector = collector;
    }

    public void close(){
    }

    public void nextTuple(){
       Utils.sleep(100L);
        String words[] = {
        "nathan","mike","jackson","golda","bertels"
        };

    public void ack(Object obj){
    }

    public void fail(Object obj){
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer){
        declarer.declarer(new Fields(new String[] {
            "word"
    }));
    }

    public Map getComponentConfiguration(){
        if(!_isDistributed_)
        {
            Map ret = new HashMap();
            ret.put(Config.TOPOLOGY_MAX_TASK_PARALLELISM,Integer.valueOf(1));
}else{
        return null;
        }
    }
}

1、类中有对WordCount的两个重载的构造函数,其中_isDistributed指明了Spout的并行度,若_isDistributed=false,则意味着这个Spout运行时仅有一份任务实例。

2、open()函数的实现,将传入的collector赋值给局部变量,使之后通过该局部变量来操作数据项的发送。

3、declareOutputFields()函数,生命了输出流的数据项结构。

4、nextTuple函数,让一只执行的线程休眠100毫秒,再继续执行下述函数体,通过线程的休眠,控制nextTuple()产生数据项的周期为0.1秒。并且在维护字符串数组中,随机挑选一个字符串,作为"word"的域,交给变量collector作为一个Tuple发送。 (ack的作用是确认数据项是否被完整处理,这里没做处理)

5、getComponentConfiguration()函数则返回组建的配置信息(这个实例中只有在_isDistributed=false时,才返回包含该配置项的Map数据结构。

6、其他重载函数都为空实现。

那么在Topology实现类的main函数使其作为一个spout:

TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout("sentenceGenSpout",new WordCount());

 

目录
相关文章
|
网络协议 算法 Java
|
存储 监控 固态存储
在高并发环境下,如何优化 WAL 的写入性能?
在高并发环境下,如何优化 WAL 的写入性能?
224 2
|
数据采集 存储 XML
Python实现网络爬虫自动化:从基础到实践
本文将介绍如何使用Python编写网络爬虫,从最基础的请求与解析,到自动化爬取并处理复杂数据。我们将通过实例展示如何抓取网页内容、解析数据、处理图片文件等常用爬虫任务。
1784 1
|
存储 缓存 监控
Flink如何优化?需要注意哪些方面?
【10月更文挑战第10天】Flink如何优化?需要注意哪些方面?
470 6
|
安全 数据安全/隐私保护
HTTPS 的实现方式
随着技术的不断发展和安全威胁的变化,HTTPS 的实现也在不断演进和完善,以提供更可靠的安全保障。通过合理运用这些实现方式,可以确保 HTTPS 在网络通信中发挥重要的安全作用,保护用户的隐私和数据安全
1746 68
|
人工智能 自然语言处理 PyTorch
Text2Video Huggingface Pipeline 文生视频接口和文生视频论文API
文生视频是AI领域热点,很多文生视频的大模型都是基于 Huggingface的 diffusers的text to video的pipeline来开发。国内外也有非常多的优秀产品如Runway AI、Pika AI 、可灵King AI、通义千问、智谱的文生视频模型等等。为了方便调用,这篇博客也尝试了使用 PyPI的text2video的python库的Wrapper类进行调用,下面会给大家介绍一下Huggingface Text to Video Pipeline的调用方式以及使用通用的text2video的python库调用方式。
|
JavaScript
在Vue3+ElementPlus项目中使用具有懒加载的el-tree树形控件
在Vue 3和Element Plus项目中实现具有懒加载功能的el-tree树形控件,以优化大数据量时的页面性能。
2922 0
|
存储 Java 定位技术
【Java】根据IP地址获取省市
【Java】根据IP地址获取省市
1144 1
|
消息中间件 Oracle 关系型数据库
实时计算 Flink版操作报错合集之一直无法正常运行,并且网络状况良好,是什么原因导致的
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
249 8
|
Shell