使Flink SQL Kafka Source支持獨立設定并行度

前言

社區在Flink 1.12版本通過FLIP-146提出了增強Flink SQL DynamicTableSource/Sink介面的動議，其中的一個主要作業就是讓它們支持獨立設定并行度，很多Sink都已經可以配置sink.parallelism引數(見FLINK-19937)，但Source還沒動靜，這是因為Source一直以來有兩種并行的標準，一是傳統的流式SourceFunction與批式InputFormat，二是原生支持流批一體的FLIP-27 Source，并且Connector之間的實作并不統一，

筆者最近在Flink釘群閑逛時，經常看到如下圖所示的發言，可見大家對Source(主要是Kafka Source)支持獨立設定并行度的需求比較急切，

本文就來基于1.13.0版本實作該需求，注意此版本的SQL Kafka Source尚未遷移到FLIP-27，這項改進已經過驗證，可以在生產環境使用，但仍屬于過渡方案，故不會向社區發起PR，

實作ParallelismProvider

ScanTableSource的運行時邏輯需要由ScanTableSource.ScanRuntimeProvider來提供，一共有5種，如下圖所示，

顯然我們要修改SourceFunctionProvider，讓它實作FLIP-146定義的ParallelismProvider介面，表示它支持獨立設定并行度，代碼很簡單：

@PublicEvolving
public interface SourceFunctionProvider extends ScanTableSource.ScanRuntimeProvider, ParallelismProvider {

    /** Helper method for creating a static provider. */
    static SourceFunctionProvider of(SourceFunction<RowData> sourceFunction, boolean isBounded) {
        return new SourceFunctionProvider() {
            @Override
            public SourceFunction<RowData> createSourceFunction() {
                return sourceFunction;
            }

            @Override
            public boolean isBounded() {
                return isBounded;
            }
        };
    }

    /** Helper method for creating a static provider with a provided parallelism. */
    static SourceFunctionProvider of(SourceFunction<RowData> sourceFunction, boolean isBounded, Integer sourceParallelism) {
        return new SourceFunctionProvider() {
            @Override
            public SourceFunction<RowData> createSourceFunction() {
                return sourceFunction;
            }

            @Override
            public boolean isBounded() {
                return isBounded;
            }

            @Override
            public Optional<Integer> getParallelism() {
                return Optional.ofNullable(sourceParallelism);
            }
        };
    }

    /** Creates a {@link SourceFunction} instance. */
    SourceFunction<RowData> createSourceFunction();
}

添加scan.parallelism引數

在o.a.f.table.factories.FactoryUtil中添加：

public static final ConfigOption<Integer> SCAN_PARALLELISM =
        ConfigOptions.key("scan.parallelism")
                .intType()
                .noDefaultValue()
                .withDescription(
                        "Defines a custom parallelism for the scan source. "
                                + "By default, if this option is not defined, the planner will derive the parallelism "
                                + "for each statement individually by also considering the global configuration.");

修改Kafka Connector

首先修改KafkaDynamicSource：

• 在構造方法中添加@Nullable Integer parallelism及相關的代碼；
• getScanRuntimeProvider()方法的最后：

return SourceFunctionProvider.of(kafkaConsumer, false, parallelism);

• 在copy() / equals() / hashCode()方法內加上parallelism，

然后修改KafkaDynamicTableFactory，加入SCAN_PARALLELISM引數，以及使用帶并行度的KafkaDynamicSource構造方法，不再贅述，

修改Source物理執行節點

負責使ScanTableSource發揮作用的物理執行節點為CommonExecTableSourceScan，注意到它的translateToPlanInternal()方法中，對不同型別的ScanRuntimeProvider分別做了處理，我們找到SourceFunctionProvider對應的那個判斷分支，加上與并行度相關的代碼，

if (provider instanceof SourceFunctionProvider) {
    SourceFunction<RowData> sourceFunction =
            ((SourceFunctionProvider) provider).createSourceFunction();
    DataStreamSource<RowData> streamSource = env.addSource(
            sourceFunction, operatorName, outputTypeInfo);
    
    final int confParallelism = streamSource.getParallelism();
    final int sourceParallelism = deriveSourceParallelism(
            (ParallelismProvider) provider, confParallelism);
    
    Transformation<RowData> transformation = streamSource.getTransformation();
    transformation.setParallelism(sourceParallelism);
    return transformation;
}

private int deriveSourceParallelism(                                                     
        ParallelismProvider parallelismProvider, int confParallelism) {                  
    final Optional<Integer> parallelismOptional = parallelismProvider.getParallelism();  
    if (parallelismOptional.isPresent()) {                                               
        int sourceParallelism = parallelismOptional.get();                               
        if (sourceParallelism <= 0) {                                                    
            throw new TableException(                                                    
                    String.format(                                                       
                            "Table: %s configured source parallelism: "                  
                                    + "%s should not be less than zero or equal to zero",
                            tableSourceSpec.getObjectIdentifier().asSummaryString(),     
                            sourceParallelism));                                         
        }                                                                                
        return sourceParallelism;                                                        
    } else {                                                                                                  
        return confParallelism;                                                          
    }                                                                                    
}

大功告成？

將全域并行度設為10，用一條簡單的SQL陳述句測驗一下：

SELECT siteId, COUNT(orderId)
FROM rtdw_dwd.kafka_order_done_log /*+ OPTIONS('scan.parallelism'='5') */
WHERE mainSiteId = 10029
GROUP BY siteId;

emm，看起來似乎不太對，為什么Source后面的Calc節點并行度也變成了5？這是因為Calc的并行度默認以輸入流的并行度決定，所以我們還要提供強制打斷算子鏈的選項，讓Calc能夠恢復全域并行度，

在ExecutionConfigOptions中加入一個引數table.exec.source.force-break-chain：

@Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING)
public static final ConfigOption<Boolean> TABLE_EXEC_SOURCE_FORCE_BREAK_CHAIN =
        key("table.exec.source.force-break-chain")
                .booleanType()
                .defaultValue(false)
                .withDescription(
                        "Indicates whether to forcefully break the operator chain after the source.");

然后在上面改過的CommonExecTableSourceScan代碼中，加入對此引數的判斷，如果為true，則呼叫disableChaining()方法斷鏈，

final Configuration config = planner.getTableConfig().getConfiguration();
if (config.get(ExecutionConfigOptions.TABLE_EXEC_SOURCE_FORCE_BREAK_CHAIN)) {
    streamSource.disableChaining();
}

最后不要忘了修改CommonExecCalc，如果它的輸入是CommonExecTableSourceScan且上述引數生效，那么就將它的并行度直接置為PARALLELISM_DEFAULT，即全域并行度，

@Override
protected Transformation<RowData> translateToPlanInternal(PlannerBase planner) {
    final ExecEdge inputEdge = getInputEdges().get(0);
    final Transformation<RowData> inputTransform =
            (Transformation<RowData>) inputEdge.translateToPlan(planner);
    final CodeGeneratorContext ctx = /* ... */;
    final CodeGenOperatorFactory<RowData> substituteStreamOperator = /* ... */;

    int parallelism = inputTransform.getParallelism();
    if (inputEdge.getSource() instanceof CommonExecTableSourceScan) {
        final Configuration config = planner.getTableConfig().getConfiguration();
        if (config.get(ExecutionConfigOptions.TABLE_EXEC_SOURCE_FORCE_BREAK_CHAIN)) {
            parallelism = ExecutionConfig.PARALLELISM_DEFAULT;
        }
    }
    return new OneInputTransformation<>(
            inputTransform,
            getDescription(),
            substituteStreamOperator,
            InternalTypeInfo.of(getOutputType()),
            parallelism);
}

再試一試，結果符合預期：

提供強制斷鏈的引數還有一重好處，即能夠在SQL作業并行度變化時安全地恢復現場，舉個例子，若Source并行度和全域并行度起初都是5，但是在作業運行程序中發現下游處理速度不夠，而將全域并行度提升到10的話，那么原有的checkpoint將無法使用——因為并行度的變化導致了作業拓撲變化，如果我們在一開始就將table.exec.source.force-break-chain設為true，那么上面所述的情況將不會造成困擾，

The End

民那晚安晚安，

標籤：其他

上一篇：解決訪問不到Linux服務器中RabbitMQ管理頁面問題

下一篇：RabbitMQ的死信佇列和延時佇列