更新时间:2025-08-13 GMT+08:00
分享

Flink作业大小表Join能力增强

本章节适用于MRS 3.3.0及以后版本。

Flink作业大小表Join

Flink作业双流Join时存在大小表数据,通过内核broadcast策略确保小表数据发送到Join的task中,通过rebalance策略将大表数据打散到Join中,提高Flink SQL易用性,增强作业稳定性。

图1 Flink作业大小表Join

在使用Flink SQL时,该特性通过hints方法指定Join的左表或右表为广播表,另一张表为rebalance表,SQL语句示例如下,分别以A\C作为小表示例:

  • 以A表作为广播表
    • 使用Join方式
      SELECT /*+ BROADCAST(A) */ a2, b2 FROM A JOIN B ON a1 = b1
    • 使用Where方式
      SELECT /*+ BROADCAST(A) */ a2, b2 FROM A, B WHERE a1 = b1
  • 以A和C表作为广播表
    SELECT /*+ BROADCAST(A, C) */ a2, b2, c2 FROM A JOIN B ON a1 = b1 JOIN C ON a1 = c1
  • 支持通过“/*+ BROADCAST(smallTable1, smallTable2) */”方式使用该特性,兼容开源双流Join逻辑。
  • 不支持开源双流Join和该特性的切换,因为该特性会将数据广播到每个Join算子。
  • 不支持LEFT JOIN时小表为左表,RIGHT JOIN时小表为右表。

Flink作业大小表Join去重

在双流关联的业务模型中,关联算子接收到其中一个流发送的大量重复数据,则会导致下游算子需要处理大量重复数据,影响作业性能。

如A表字段(P1,A1,A2)使用如下方式关联B表字段(P1,B1,B2,B3)生成C的场景中,B表信息发生大量更新,但是B中的所需字段没有更新,在该关联中仅用到了B表的B1和B2字段,对于B表,每个记录更新只更新B3字段,B1和B2不更新,因此当B表更新,可以忽略更新后的数据。

select  A.A1,B.B1,B.B2 from A join B on A.P1=B.P1

为解决如上问题可通过使用hint单独为左表(duplicate.left)或右表(duplicate.right)设置去重:

  • 格式
    • 为左表设置去重
       /*+ OPTIONS('duplicate.left'='true')*/
    • 为右表设置去重
       /*+ OPTIONS('duplicate.right'='true')*/
    • 同时为左表和右表设置去重
       /*+ OPTIONS('duplicate.left'='true','duplicate.right'='true')*/
  • 在SQL语句中配置
    如同时为左表“user_info”和右表“user_score”设置去重。
    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
      JOIN 
      -- 为左表和右表设置去重
      user_score /*+ OPTIONS('duplicate.left'='true','duplicate.right'='true')*/ as d ON t.user_id = d.user_id;
    Kafka Broker实例IP地址及端口号说明:
    • 服务的实例IP地址可通过登录FusionInsight Manager后,单击“集群 > 服务 > Kafka > 实例”,在实例列表页面中查询。
    • 集群已启用Kerberos认证(安全模式)时Broker端口为“sasl.port”参数的值,默认为“21007”。
    • 集群未启用Kerberos认证(普通模式)时Broker端口为“port”的值,默认为“9092”。如果配置端口号为9092,则需要配置“allow.everyone.if.no.acl.found”参数为true,具体操作如下:

      登录FusionInsight Manager系统,选择“集群 > 服务 > Kafka > 配置 > 全部配置”,搜索“allow.everyone.if.no.acl.found”配置,修改参数值为true,保存配置即可。

相关文档