Slipstream中流与流、流与表之间的join

快来打我* 2021-09-18 02:20 432阅读 0赞

**1、流之间的join**

Slipstream支持Window Stream之间的join操作。Window Stream在join的时候必须保证Slide的值一致，Length的值可以不同。

这里以系统时间切分滑动窗口为例进行说明。

*  基于Kafka topic asd和zxc分别建立两个输入流s1和s2

CREATE STREAM s1 (id STRING, letter STRING)
    
      ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
    
      TBLPROPERTIES("topic"="asd",
    
      "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\""
    
      );
    
    
    
    CREATE STREAM s2 (id STRING, letter STRING)
    
      ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
    
      TBLPROPERTIES("topic"="zxc",
    
      "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\""
    
      );

*  建立两个窗口大小为10秒，滑动间隔为5秒的Window Stream

CREATE STREAM s1_join as SELECT * FROM s1 STREAMWINDOW w1 AS
    
    (LENGTH '10' SECOND SLIDE '5' SECOND);
    
    
    
    CREATE STREAM s2_join as SELECT * FROM s2 STREAMWINDOW w1 AS
    
    (LENGTH '10' SECOND SLIDE '5' SECOND);

*  建立一个结果表

CREATE TABLE t1 (id STRING, letter1 STRING, letter2 STRING);

*  触发一个join操作

INSERT INTO t1 SELECT s1_join.id, s1_join.letter, s2_join.letter FROM s1_join JOIN s2_join ON s1_join.id = s2_join.id;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70][]

*  启动kafka生产者，生产测试数据（为便于测试，topic asd和topic zxc生产的数据均如下图所示）

![20190219144913586.png][]

*  一段时间后查询t1表中的数据

![20190219144913685.png][]

**注意：**join操作不能使用无限滑动窗口。

**2、 流与表的join**

由于join操作不能使用无限滑动窗口，在使用join进行实时数据分析的时候，很难将流中的所有数据进行join。另一方面，即使可以将两个流中的所有数据进行join操作，但当数据体量巨大的时候，其实时性也会受到影响。为了在不影响实时性的情况下还可以进行join操作，可以使用流与表之间的join操作进行数据分析。

**2.1 流与表之间的MapJoin**

流与表之间的MapJoin可以使用HINT来实现。HINT是写在语句中用于指示计算底层实现某个操作执行方式的一条说明，按照优化目的与一定的格式书写。MapJoin的HINT形式为 /\*+ MAPJOIN(tables) \*/，其中tables指明被join的表，可以为多个，一般都是小表。HINT的实现语法如下：

SELECT /*+MAPJOIN(<small_table_name>)*/ [查询的字段名] FROM <stream_name> JOIN <small_table_name> on ...;

因为MapJoin开关默认开启（ngmr.mapjoin.autoconvert=TRUE），当表比较小，符合MapJoin的小表定义时（由参数hive.mapjoin.smalltable.filesize指定），流与表的join默认将被转成MapJoin模式。所以只要join的表在小表的大小范围之内，也可以用一般形式实现MapJoin。这里以一般形式为例实现流与表之间的MapJoin，HINT形式除在书写上与一般形式不同之外，实现方法与一般形式均相同。

*  MapJoin中不支持es、HBase等带索引的表，而Slipstream中不支持普通表的增删改等操作。故使用如下方法建立一个测试表。

CREATE TABLE tt (id STRING, letter STRING) stored as ES with shard number 8 replication 1;
    
    
    
    insert into tt values('1','a');
    
    insert into tt values('2','b');
    
    insert into tt values('3','c');
    
    insert into tt values('4','d');
    
    insert into tt values('5','e');
    
    insert into tt values('6','f');
    
    insert into tt values('7','g');
    
    
    
    create table tab as select * from tt;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1][]

*  创建一个输入流并新建一个窗口大小和滑动间隔为5秒的Window Stream。

CREATE STREAM s1 (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\""
    
      );
    
    
    
    CREATE STREAM s1_join as SELECT * FROM s1 STREAMWINDOW w1 AS
    
    (INTERVAL '5' SECOND);

![20190219144913589.png][]

*  创建一个衍生流，使用一般形式实现流与表的join。

CREATE STREAM s2 AS SELECT ss.id,ss.letter l1,t.letter l2 from s1_join ss join tab t on ss.id = t.id;

![20190219144913590.png][]

*  创建一个es表用于存储join后的结果。

CREATE TABLE t1 (id STRING, l1 STRING, l2 STRING) stored as ES with shard number 8 replication 1;

![20190219144913676.png][]

*  触发流.

insert into t1 select * from s2;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 2][]

*  开启生产者生产数据。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 3][]

*  查询表t1，查看join之后的数据。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 4][]

继续生产数据，查询表t1可得：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 5][]

MapJoin操作虽然实现了流与表之间的join，但是却存在着一个问题：被join表的数据必须是固定的，即如果该表中的数据是从另外一个流中实时的传输进来的，那么MapJoin操作只会对流传数据进来之前已经存在的数据进行join操作，实时传输进来的数据是无法获取到的。可以使用流与表之间的GlobalLookupJoin来解决这个问题。

**2.2 流与表之间的GlobalLookupJoin**

GlobalLookupJoin仅适用于事件驱动模式，它主要用于流和大表做关联分析，表的类型需为带索引的表，例如：es表、HBase表等。GlobalLookupJoin的形式为/\* +glkjoin(tables) \*/，其中tables是大表的名称，必须在join的右侧。可以通过设置stargate.global.lookup.join.batchsize调整join的batch大小（默认是1000），以控制吞吐和延迟之间的平衡。

*  创建一个输入流并创建用于测试的es表。

![20190219144913780.png][]

*  创建一个输入流并新建一个窗口大小和滑动间隔为5秒的Window Stream。

CREATE STREAM s2 (id STRING, letter STRING)
    
      ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
    
      TBLPROPERTIES("topic"="zxc",
    
      "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\""
    
      );
    
    
    
    CREATE STREAM s2_join as SELECT * FROM s2 STREAMWINDOW w1 AS
    
    (INTERVAL '5' SECOND);

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 6][]

*  创建一个衍生流，实现GlobalLookupJoin。

CREATE STREAM s3 AS SELECT /* +glkjoin(t) */ ss.id,ss.letter l1,t.letter l2 from s2_join ss join tab t on ss.id = t.id;

![20190219144913789.png][]

*  创建es表，用于接收join后的结果数据。

CREATE TABLE t1 (id STRING, l1 STRING, l2 STRING) stored as ES with shard number 8 replication 1;

![20190219144913821.png][]

*  触发流s1和s3。

insert into tab select * from s1;
    
    
    
    insert into t1 select * from s3;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 7][]

*  启动生产者生产数据。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 8][]

查询结果表t1中的数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 9][]

*  继续生产数据。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 10][]

查询结果表t1中的数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 11][]

流与表之间的GlobalLookupJoin解决了被join表的数据不能是可变的问题。同时，可以在join之后的stream上使用window，进行一系列聚合操作。例如：使用下列语句建立无限滑动窗口并统计某一列各个值出现的次数：

CREATE STREAM s4 AS SELECT * FROM s3 STREAMWINDOW w1 AS (LENGTH INFINITE SLIDE '10' SECOND);
    
    
    
    CREATE TABLE t2 (l1 STRING, sl STRING) stored as ES with shard number 8 replication 1;
    
    
    
    INSERT into t2 SELECT l1, count(*) as sl FROM s4 GROUP BY l1;

查询统计结果为：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 12][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70]: /images/20210726/8230750c165b408f9846f74588d11554.png
[20190219144913586.png]: /images/20210726/a6ec013c225a4e55a1de41fffb5afde0.png
[20190219144913685.png]: /images/20210726/ea23acdd0cab44ceacbfec1d96e96954.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1]: /images/20210726/a6ec013c225a4e55a1de41fffb5afde0.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM=,size_16,color_FFFFFF,t_70
[20190219144913589.png]: /images/20210726/a6846e67327e42dd9e0180d9f09e3dc9.png
[20190219144913590.png]: /images/20210726/ea443c5cf7fd46a083c6c785430fda87.png
[20190219144913676.png]: /images/20210726/68f3d3333d2e4f3db6110596051fd167.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 2]: /images/20210726/4a728b2e96634edb87091bfa9263b96d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 3]: /images/20210726/b59ed5e37b754e54b11b8744f78d0454.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 4]: /images/20210726/ed84f1df82ca483aab778831da8343da.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 5]: /images/20210726/0784d07fcd9143a58762615942af0432.png
[20190219144913780.png]: /images/20210726/9079f9fa1f14484bb4a6002e1b4654bd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 6]: /images/20210726/6aca21a42bb949e686e751b5373a05ce.png
[20190219144913789.png]: /images/20210726/147d563bcffe461b88223997c4c2563d.png
[20190219144913821.png]: /images/20210726/0217208039af4348bc18bea0128f09e6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 7]: /images/20210726/ff9f011b03c244d0a40ef13f3db676ea.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 8]: /images/20210726/d447d2fd84104a5b881549f4ad36937a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 9]: /images/20210726/b82145a7a92249ceba618ba7d153551e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 10]: /images/20210726/4664f3a6bcf045709b47723b81317656.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 11]: /images/20210726/813f9a54fc98465b9d9bd773173ce1b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 12]: /images/20210726/66675a6b6d2f43fc8583ce19ea9ba768.png