Slipstream中的衍生流（Derived Stream）
[《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中创建一个包含三个字段的测试表如下：
- 启动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监控界面查看触发是否成功：
- 测试衍生流与Flume、Kafka的对接
在Oracle中插入几条数据，如下：
查询table_c中是否已经接收到了合并NAME和PWD字段的数据：
至此，衍生流对接Flume、Kafka成功。
不念不忘少年蓝@
Dear 丶
╰+哭是因爲堅強的太久メ
傷城~
还没有评论，来说两句吧...