Slipstream中的衍生流（Derived Stream）

╰+哭是因爲堅強的太久メ 2021-09-18 00:30 323阅读 0赞

[《Flume、Kafka、Slipstream对接》][Flume_Kafka_Slipstream]中介绍的流属于输入流，一般原始数据并不是最终要输出的数据，它需要经过过滤，提炼，计算。这样一个过程称之为对Stream的转化。可以采用CSAS(CREATE STREAM AS SELECT)对已有的Stream进行变形。转化得到的新的Stream，称之为Derived Stream。

**1、衍生流的建立**

使用如下语句首先建立一个输入流：

CREATE STREAM stream_a(id STRING, name STRING, pwd STRING)
    
      ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
    
      TBLPROPERTIES("topic"="yxy",
    
      "kafka.zookeeper"="node3:2181,node2:2181,node1:2181",
    
      "kafka.broker.list"="node3:9092,node2:9092,node1:9092",
    
      "transwarp.consumer.security.protocol"="SASL_PLAINTEXT",
    
      "transwarp.consumer.sasl.mechanism"="GSSAPI",
    
      "transwarp.consumer.sasl.kerberos.service.name"="kafka",
    
      "transwarp.consumer.sasl.jaas.config"="com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab=\"/etc/slipstream1/conf/kafka.keytab\" principal=\"kafka@TDH\""
    
      );

在输入流的基础上建立一个衍生流（衍生流可以建立任意多个，这里以一个为例，对NAME和PWD字段做合并）：

create stream stream_b as select id,name||pwd as np from stream_a;

**2、衍生流与Flume、Kafka的对接**

*  创建Oracle测试表

在Oracle中创建一个包含三个字段的测试表如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70][]

*  启动Flume

使用如下命令启动Flume组件：

nohup bin/flume-ng agent --name a1 --conf ./conf --conf-file ./conf/flume-topic-oracle.conf -Dflume.root.logger=INFO,console -Djava.security.auth.login.config=/opt/flume/apache-flume-1.7.0-bin/conf/jaas.conf &

*  触发流

首先按照1中先创建一个输入流和一个衍生流，然后创建一个表，例如：

CREATE TABLE table_c (id STRING, np STRING);

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

INSERT INTO table_c SELECT * FROM stream_b;

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1][]

*  测试衍生流与Flume、Kafka的对接

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 2][]

查询table\_c中是否已经接收到了合并NAME和PWD字段的数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 3][]

至此，衍生流对接Flume、Kafka成功。

[Flume_Kafka_Slipstream]: https://blog.csdn.net/gdkyxy2013/article/details/86741961
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70]: /images/20210726/32ebb97b02864a4c8a42270e43aaf15a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1]: /images/20210726/c1eb229f95964877993d6fa5605c5a3d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 2]: /images/20210726/c00b6cd957ba4ba5bcdf376a8b0786cb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 3]: /images/20210726/bf8485f612594b62bf590cd81d6a2314.png