PAI-Rec系统中流量调控的使用流程

准备工作

在具体创建流量调控任务前，需要做以下准备工作（该配置做一次即可）：

1. 下载流量调控JAR包（流量调控JAR包）和 FeatureStore 的JAR包 ，具体版本见设置Flink Connector文档。

2. 进入Flink实时计算控制台，如果没有购买实例，需要先购买。如果已有购买好的，选择工作空间（注意，需要让主账户或者购买实例的用户把你加入到工作空间，否则没有权限进入），进入工作空间之后，左侧导航栏选择连接器，然后点击创建自定义连接器，上传已经下载的Jar包，就可以注册 Connector了。流量调控的Jar包中包含多个连接器（如traffic-ctrl-sink、traffic-ctrl-lookup、traffic-lookup、http-sink、rest-lookup），需要逐个上传并创建。如果 Connector 有 Lookup 选项，就勾选上，其他的参数不要改动。

3. 点击左侧目录中 ETL => 函数 => 点击+号，还是上传刚才下载好的流量调控Jar包去注册UDF。在注册UDF JAR 的窗口中，我们约定UDF名称填写为 traffic-ctrl ，最后点击确定。系统扫描JAR之后，会显示UDF函数列表是：ConditionMatch、DynamicTarget、ExplodeSplit，确认创建函数。

PAI-Rec侧

1. 新建流量调控任务

回到PAI-Rec控制台，左侧导航栏找到运营工具->流量调控，点击新增任务。

1.1 设置基础信息

新建流量调控任务时，需依次填写任务名称、任务描述，选择生效场景和引擎服务，配置Flink数据源和执行时间（指定时间段或永久执行），并在依赖数据源配置中选择行为表、物品表和用户表对应的数据表。页面底部还需选择预发实验和生产实验对应的实验组和实验。

1.2 圈选干预人群

自定义表达式支持多种运算符，包含数学运算符、逻辑运算符、比较运算符及集合运算符等。

• 数学运算符包括加（+）、减（-）、乘（*）、除（/）和取模（%）。

• 逻辑运算符包括逻辑与(&&)、逻辑或(||)和逻辑非(!)。

• 比较运算符有相等（==）、不等（!=）、小于（<）、小于等于（<=）、大于（>）和大于等于（>=）。

• 操作集合的运算符，如in用于检查元素是否在集合中。

1.3 设置调控目标

调控目标：一个调控任务下，可以包含多个调控目标，调控目标是真正圈选调控物品集并设置目标的

配置完之后，点击保存，会进入发布页面。

1.4 发布Flink任务

此处会生成两段 Flink Sql 代码，

Flink任务的目的是实时统计被调控目标的数量。

1. 重置datahub/kafka的行为数据的点位到当天的0点（重置点位旨在校准消费起点，以消除历史状态干扰，确保数据在时间窗口和处理逻辑上的严格对齐，从而保障统计结果的完整与准确。）

   发布任务时可能出现提示弹窗，请按照界面文档提示完成Datahub配置及Debug配置。

   弹窗中显示Datahub配置和Debug配置的完成状态，单击前往写配置按钮可跳转至对应配置页面完成设置。

2. 生成的统计SQL在运行时需要用到ak信息，首次创建流控任务需要点击创建Flink项目变量，然后在发布任务前需要前往Flink平台配置您的ak信息。页面中红色提示区域显示需要配置ACCESS_KEY_ID和ACCESS_KEY_SECRET两个变量。

   提示创建变量成功之后，在Flink控制台变量管理会新增以下两个变量，请设置您的AK信息。

   这两个变量分别为ACCESS_KEY_ID和ACCESS_KEY_SECRET，类型均为密文，请在此处填入您的AK凭证。

3. 以上做完后点击发布Flink任务即可（预发环境发布完成，生产环境方可发布）。 发布需要等待一段时间，可点击查看Flink发布状态观测具体发布流程，在Flink平台ETL=>作业草稿可以看到对应的SQL草稿（后缀为任务ID）以及作业运维中当前作业运行情况（启动中=>运行中）。

   在流量调控任务页面，切换到预发环境，单击发布Flink任务按钮。系统提示「已触发发布，部署中需稍等3min再查看」。任务列表中可查看各Flink作业的部署状态。发布Flink任务成功如下：

   在Flink平台ETL > 作业运维中，可以看到已部署的Flink作业，状态为RUNNING表示任务正常运行。

   在任务列表中，可通过启动开关控制流量调控任务的启停状态。

   发布Flink任务操作，可能会出现err，您根据报错提示调整后可点击重试按钮重新执行后续阶段流程或去前往Flink平台详细排查原因并部署启动对应Flink作业。

2. 发布/启动任务

发布调控任务之后，任务参数不可以再调整，只能新增或者关闭调控目标。

发布完之后，才可以进行启动，先启动预发流量调控任务，整个链路运行正常后，再启动生产环境任务。

在左上角可以切换环境，在这里可以发布或者启动不同环境的任务。

引擎侧

在原来的引擎配置上，根据业务情况，添加下面的配置。

1.流量调控配置

该配置在预发环境和预发环境上都要配置。

• DefaultKp, DefaultKi, DefaultKd 为 PID 算法的三个默认参数值：Kp,Ki,Kd，可通过A/B实验配置覆盖。

• DiversityRuleSort 为多样性重排（DiversityRuleSort）。

    "SortConfs": [
        {
            "Name": "TrafficControlSort",
            "SortType": "TrafficControlSort",
            "PIDConf": {
                "DefaultKp": 5,
                "DefaultKi": 1,
                "DefaultKd": 1
            }
        },
        {
            "Name": "TrafficDiversityRuleSort",
            "SortType": "DiversityRuleSort",
            "DiversitySize": 20,
            "DiversityRules": [
                {
                    "Dimensions": [
                        "__traffic_control_id__"
                    ],
                    "WindowSize": 10,
                    "FrequencySize": 2
                }
            ]
        }
    ],
    "SortNames": {
        "home_feed": [
            "TrafficControlSort",
            "TrafficDiversityRuleSort"
        ]
    },

• TrafficControlSort 是引擎中流量调控的模块名，这个模块执行后，会给每个item添加一个特殊字段：__traffic_control_id__。

  • __traffic_control_id__ 的含义 ：标记当前item的位置是否被“向前移动”，如果被“向前移动”，则该字段的值固定为 0，否则被赋值为它在列表中的原本的位置。

  • 上面向前移动的概念为：一个item数组，本来某个 item 在请求的size之后，意味着本来没有曝光的机会，如果流量调控把这个item提到了 size 之内，有了曝光机会，这种情况才会被标记为 __traffic_control_id__=0，其他情况，如果某个 item 本来就在 size 之内，调控之后还是 size 之内，是不会标记为 0 的，因为本来就在曝光列表中，调控之后，还在曝光列表中，这种会认为是自然流量，不是流量调控的功劳。

• DiversityRuleSort 是引擎中打散规则的模块名，可以针对__traffic_control_id__这个字段配置打散约束，比如输出列表中， 每10个位置中最多只能出2个被调控“向前移动”的item。

  • 如果没有DiversityRuleSort，那么很有可能用户满屏看到的都是被调控后，从排序靠后的位置移到前面的item，这些item的相关性一般比较弱，导致系统失去个性化，对用户体验有一定的损伤，严重影响大盘指标。

2.进阶配置

2.1 按比例调控

无额外配置。

2.2 按量\次数调控

该配置在预发环境和预发环境上都要配置，配置完流量调控后，还需添加下面的配置。直接召回所有的被调控物品会给排序造成很大压力，所以需要自定义一个PipelineConfs，每次请求从召回表里取RetainNum：300个物品，进行排序，然后合并到主pipeline里，配置详情参考自定义 Pipeline 流程。

"PipelineConfs": {
    "home_feed": [
        {
            "Name": "newitem",
            "FilterNames": [
                "UniqueFilter",
                "UserExposureFilter15Min",
                "UserExposureFilterReal",
                "adjust_count_filter"
            ],
            "RecallNames": [
                "new_item_1500_recall"
            ],
            "RankConf": {
                "RankAlgoList": [
                    "fs_dbmtl_v4"
                ],
                "RankScore": "${fs_dbmtl_v4_probs_is_click}+${fs_dbmtl_v4_probs_is_collect_like_comment}",
                "BatchCount": 400,
                    "Processor": "EasyRec"
                },
                FeatureLoadConfs": [
                    {
                        "FeatureDaoConf": {
                        "AdapterType": "hologres",
                        "HologresName": "holo_info",
                        "ItemFeatureKeyName": "item_id",
                        "FeatureKey": "item:id",
                        "HologresTableName": "xxx_home_feed_new_item_1500_v2_online",
                        "ItemSelectFields": "item_id,is_1500_new_item",
                        "FeatureStore": "item"
                    },
                    "Features": []
                }
            ]
        }
    ]
},
"RecallConfs": [
    {
        "Name": "new_item_1500_recall",
        "RecallType": "ColdStartRecall",
        "RecallCount": 1500,
        "ColdStartDaoConf": {
            "AdapterType": "hologres",
            "HologresName": "holo_info",
            "HologresTableName": "xxx_home_feed_new_item_1500_v2_online",
            "WhereClause": "is_complated='1'",
            "PrimaryKey": "\"item_id\""
        }
    }
],
"FilterConfs": [
    {
        "Name": "adjust_count_filter",
        "FilterType": "AdjustCountFilter",
        "ShuffleItem": true,
        "RetainNum": 300
    }
]