flink 源码调试flinksql的 explain功能

水深无声 2022-11-17 13:48 197阅读 0赞

1、因业务需要，需要把查看sql的执行计划，所以在代码里可以添加如下代码来查看相关数据,在  **TableEnvironmentITCase**测试类添加代码

//todo 查看sql的解析语法
      @Test
      def testExplainAndExecuteSingleSink2(): Unit = {
        val sinkPath = TestTableSourceSinks.createCsvTemporarySinkTable(
          tEnv, new TableSchema(Array("first","totalQuantity"), Array(STRING,DOUBLE)),"MySink1")
    
        //    val table1 = tEnv.sqlQuery("select first from MyTable")
        val table1 = tEnv.sqlQuery(
          "SELECT first,totalQuantity FROM (  " +
            "SELECT first,totalQuantity,row_number() OVER(PARTITION BY first ORDER BY totalQuantity DESC) AS rownum  FROM (  " +
            "  SELECT first , sum(score) AS totalQuantity " +
            " FROM MyTable GROUP BY first  )" +
            ") WHERE rownum <= 10")
        
        tEnv.insertInto(table1, "MySink1")
        //    todo 打印相关日志信息
        println(tEnv.explain(false))
        tEnv.execute("test1")
      }

2、debug日志如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dwcHdwcDE_size_16_color_FFFFFF_t_70][]

3、ast语法树

== Abstract Syntax Tree ==
    LogicalLegacySink(name=[`default_catalog`.`default_database`.`MySink1`], fields=[first, totalQuantity])
    +- LogicalProject(first=[$0], totalQuantity=[$1])
       +- LogicalFilter(condition=[<=($2, 10)])
          +- LogicalProject(first=[$0], totalQuantity=[$1], rownum=[ROW_NUMBER() OVER (PARTITION BY $0 ORDER BY $1 DESC NULLS LAST)])
             +- LogicalAggregate(group=[{0}], totalQuantity=[SUM($1)])
                +- LogicalProject(first=[$0], score=[$2])
                   +- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [CsvTableSource(read fields: first, id, score, last)]]])
    
    == Optimized Logical Plan ==
    LegacySink(name=[`default_catalog`.`default_database`.`MySink1`], fields=[first, totalQuantity])
    +- Calc(select=[first, totalQuantity], where=[<=(w0$o0, 10)])
       +- OverAggregate(partitionBy=[first], orderBy=[totalQuantity DESC], window#0=[ROW_NUMBER(*) AS w0$o0 ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], select=[first, totalQuantity, w0$o0])
          +- Sort(orderBy=[first ASC, totalQuantity DESC])
             +- HashAggregate(isMerge=[true], groupBy=[first], select=[first, Final_SUM(sum$0) AS totalQuantity])
                +- Exchange(distribution=[hash[first]])
                   +- LocalHashAggregate(groupBy=[first], select=[first, Partial_SUM(score) AS sum$0])
                      +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [CsvTableSource(read fields: first, score)]]], fields=[first, score])
    
    == Physical Execution Plan ==
    Stage 14 : Data Source
    	content : Source: Custom File source
    
    	Stage 15 : Operator
    		content : CsvTableSource(read fields: first, score)
    		ship_strategy : REBALANCE
    
    		Stage 16 : Operator
    			content : SourceConversion(table=[default_catalog.default_database.MyTable, source: [CsvTableSource(read fields: first, score)]], fields=[first, score])
    			ship_strategy : FORWARD
    
    			Stage 17 : Operator
    				content : LocalHashAggregate(groupBy=[first], select=[first, Partial_SUM(score) AS sum$0])
    				ship_strategy : FORWARD
    
    				Stage 19 : Operator
    					content : HashAggregate(isMerge=[true], groupBy=[first], select=[first, Final_SUM(sum$0) AS totalQuantity])
    					ship_strategy : HASH[first]
    
    					Stage 20 : Operator
    						content : Sort(orderBy=[first ASC, totalQuantity DESC])
    						ship_strategy : FORWARD
    
    						Stage 21 : Operator
    							content : OverAggregate(partitionBy=[first], orderBy=[totalQuantity DESC], window#0=[ROW_NUMBER(*) AS w0$o0 ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], select=[first, totalQuantity, w0$o0])
    							ship_strategy : FORWARD
    
    							Stage 22 : Operator
    								content : Calc(select=[first, totalQuantity], where=[(w0$o0 <= 10)])
    								ship_strategy : FORWARD
    
    								Stage 23 : Operator
    									content : SinkConversionToRow
    									ship_strategy : FORWARD
    
    									Stage 24 : Operator
    										content : Map
    										ship_strategy : REBALANCE
    
    										Stage 25 : Data Sink
    											content : Sink: CsvTableSink(first, totalQuantity)
    											ship_strategy : FORWARD

4、

VolcanoPlanner优化器

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dwcHdwcDE_size_16_color_FFFFFF_t_70 1][]

VolcanoCost

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dwcHdwcDE_size_16_color_FFFFFF_t_70]: /images/20221022/1f97b55bc10549a2bedf89bd961c649e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dwcHdwcDE_size_16_color_FFFFFF_t_70 1]: /images/20221022/ad948986bb134bf7a2e38b5c0895c3d3.png