Slipstream中的并行流处理

淡淡的烟草味﹌ 2021-09-18 00:32 499阅读 0赞 
  1.    与[《Kafka中的消费者组》][Kafka]类似，Slipstream中多个流也可以同时接收同一topic的数据进行不同的操作。值得注意的是，Slipstream中的一个Input Stream对应着Kafka中的一个Consumer Group。

1、并行流的建立

  1.    并行流的建立非常简单，在[《Slipstream中的衍生流》][Slipstream]建立的基础上，再建立一个新的输入流和衍生流即可，新建的输入流需与之前的输入流指向同一个topic。为了测试并行，这里指定新建的衍生流增加一个字符串截取的功能，如下：
  3. create stream stream_b2 as select id,substr(name||pwd,1,4) as sp from stream_a2;

其中，stream_a2为新建的输入流。

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70

2、并行流处理数据

  • 创建Oracle测试表并启动Flume

此操作与《Slipstream中的衍生流》中创建Oracle测试表、启动Flume的操作相同。

  • 触发流

创建一个表，用于接收stream_b2传过来的数据，例如：

  1. CREATE TABLE table_c2 (id STRING, sp STRING);

最后，通过下列语句触发流：

  1. INSERT INTO table_c SELECT * FROM stream_b;
  3. INSERT INTO table_c2 SELECT * FROM stream_b2;

通过4044监控界面查看触发是否成功：

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1

  • 测试并行流

在Oracle中插入几条数据，如下：

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 2

查询table_c和table_c2中是否接收到了处理后的全部数据：

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 3

至此，并行流测试成功。

发表评论

表情：
评论列表 （有 0 条评论，499人围观）

还没有评论，来说两句吧...

相关阅读

    相关 并行流parallelStream

    1.并发与并行的区别： 并发： 一个时间段内有几个程序都处于已启动到运行完毕之间，且这几个程序都是在同一个处理机上运行。但在任一个时刻点只有一个程序在处理机上运行

    ╰+攻爆jí腚メ╰+攻爆jí腚メ/ 0 赞/ 396 阅读

    相关 Slipstream中的高可用（HA）

           一个Application或者一个StreamJob，如果上游的流发生故障（例如意外退出）无法及时恢复，可能会导致整个系统的瘫痪。因此，流处理系统的高可用性显得尤

    傷城~傷城~/ 0 赞/ 591 阅读