# FlinkSQL支持双流Join过滤Join产生的回撤
#### 使用场景
在Flink双流Join场景中，当其中一流的数据发生更新时，Join算子会产生回撤消息（-D和-U），这些回撤消息发送到下游后，可能导致下游算子产生大量中间结果，增加计算开销并影响数据准确性。配置Flink作业时，双流Join可以过滤Join产生的回撤，source算子的-D +U -U不会过滤。
#### 约束与限制
- 只在双流join时使用。
- 不推荐在join后有其他算子，join之后直接输出到sink。
- 支持inner join、left join和right join，不支持MultiJoin算子。
- 本章节仅适用于MRS 3.6.0-LTS及之后版本。
 
#### 使用方法
在join时添加hints参数 /\*+ OPTIONS('allows.generated-retracts'='false') \*/。
SQL示例：
```
CREATE TABLE user_info (`user_id` VARCHAR, `user_name` VARCHAR) WITH (
  'connector' = 'kafka',
  'topic' = 'user_info_001',
  'properties.bootstrap.servers' = 'Kafka的Broker实例业务IP:Kafka端口号',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'latest-offset',
  'value.format' = 'csv'
);
CREATE table print(
  `user_id` VARCHAR,
  `user_name` VARCHAR,
  `score` INT
) WITH ('connector' = 'print');
CREATE TABLE user_score (user_id VARCHAR, score INT) WITH (
  'connector' = 'kafka',
  'topic' = 'user_score_001',
  'properties.bootstrap.servers' = 'Kafka的Broker实例业务IP:Kafka端口号',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'latest-offset',
  'value.format' = 'csv'
);
INSERT INTO
  print
SELECT
  t.user_id,
  t.user_name,
  d.score
FROM
  user_info as t
  LEFT JOIN 
  /*+ OPTIONS('allows.generated-retracts'='false') */ 
  user_score as d ON t.user_id = d.user_id;
```
![](https://support.huaweicloud.com/cmpntguide-lts-mrs/public_sys-resources/note_3.0-zh-cn.png)
Kafka Broker实例IP地址及端口号说明：
- 服务的实例IP地址可通过登录Manager后，单击"集群 \> 服务 \> Kafka \> 实例"，在实例列表页面中查询。 登录集群Manager具体操作，请参考[访问MRS集群Manager](https://support.huaweicloud.com/usermanual-mrs/mrs_01_0128.html)。
  
- 集群已启用Kerberos认证（安全模式）时Broker端口为"sasl.port"参数的值，默认为"21007"。
- 集群未启用Kerberos认证（普通模式）时Broker端口为"port"的值，默认为"9092"。如果配置端口号为9092，则需要配置"allow.everyone.if.no.acl.found"参数为"true"，具体操作如下： 登录Manager系统，选择"集群 \> 服务 \> Kafka \> 配置 \> 全部配置"，搜索"allow.everyone.if.no.acl.found"配置，修改参数值为"true"，保存配置即可。
  
 
数据示例：
user_info表先输入数据\["1"，"zs"\]
user_score表后输入数据\["1"，25\]
print sink打印结果：
- 不使用特性时 +I\["1"，"zs"，null\]
  -D\["1"，"zs"，null\]
  +I\["1"，"zs"，25\]
  
 
- 使用特性时 +I\["1"，"zs"，null\]
  +I\["1"，"zs"，25\]