Flume、Kafka、Slipstream对接

       在启动Slipstream之前首先去到8180监控界面，查看Slipstream的server在node2节点上，如下图所示：

        Slipstream和Inceptor的server不在一个节点上，只能共同使用同一个源数据库，相关操作只能在Slipstream的引擎中操作。

        安全模式下，一个Stream要成功读取/写入一个topic中的消息必须满足两个条件：

       （1）每个运行Executor的服务器上都必须有建stream用户的keytab。

        首先需要下载keytab:8180监控界面à管理àGuardianà角色àGuardian ServeràLinkà登录到8380界面（也可在浏览器中直接输入地址跳转）。

         登录到8380界面之后，点击“租户”à选择kafkaà点击“下载keytab”

          将下载后的keytab重命名为kafka.keytab并将其上传到node1的/etc/slipstream1/conf目录下，使用如下命令分发到其余各个节点：

scp -r /etc/slipstream1/conf/kafka.keytab node2: /etc/slipstream1/conf/

      （2）建Stream的用户有该topic的读/写权限（SreamJob落地的表格式如果为hyperdrive表，因为创建Stream的用户是hive，所以也要确保hive用户在hbase中的权限）。

1、进入Slipstream

使用如下命令进入Slipstream：

beeline -u 'jdbc:hive2://node2:10010/default' -n hive -p password

如下图所示：

2、Flume、Kafka、Slipstream对接

        在《Flume抽取Oracle中的数据到Kafka》中，Flume已经成功对接Kafka。这里只需要将Kafka和Slipstream对接即可。

2.1 建流

使用如下的语句建立一个流：

CREATE STREAM aaa (id STRING, letter 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""

  );

         在TBLPROPERTIES中可以指定Kafka的topic等相关设置。执行show streams;可以查看已经建立的流。

2.2 建立与流相对应的表

        此处需要注意的是表的字段及类型必须与流的一致，如下：

CREATE TABLE bbb (id STRING, letter STRING);

       使用show tables;命令可以查看所创建的所有流和表。

2.3 触发流

       Flume从Oracle数据库中抽取的数据经过Kafka到达Slipstream中，要想使用Slipstream中的数据需将数据插入到表中。使用如下的命令触发流，使流中的数据实时的导入到Inceptor表中。

INSERT INTO bbb SELECT * FROM aaa;

       可以到4044界面查看，触发流操作是否成功。如下图所示：

2.4 测试Flume、Kafka、Slipstream的对接

       向Oracle对应表中插入几条数据，查看Inceptor对应表中是否接收到数据：

      至此，说明三者对接成功。

3、停止StreamJob

      使用如下命令列出正在运行的StreamJob：

LIST STREAMJOBS;

     使用如下命令停止一个StreamJob：

STOP STREAMJOB streamid;

     如下图所示：

最后

以上就是复杂期待为你收集整理的Flume、Kafka、Slipstream对接的全部内容，希望文章能够帮你解决Flume、Kafka、Slipstream对接所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。