【源码解析】flink sql执行源码概述:flink sql执行过程中有哪些阶段,这些阶段的源码大概位置在哪里

这篇具有很好参考价值的文章主要介绍了【源码解析】flink sql执行源码概述:flink sql执行过程中有哪些阶段,这些阶段的源码大概位置在哪里。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

本文大致分析了flink sql执行过程中的各个阶段的源码逻辑,这样可以在flink sql执行过程中, 能够定位到任务执行的某个阶段的代码大概分布在哪里,为更针对性的分析此阶段的细节逻辑打下基础,比如create 的逻辑是怎么执行的,select的逻辑是怎么生成的,优化逻辑都做了哪些,而这些是接下来的文章要分析的。

一. sql执行流程源码分析

SQL语句经过Calcite解析生成抽象语法树SQLNode,基于生成的SQLNode并结合flink Catalog完成校验生成一颗Operation树,接下来blink planner将Opearation树转为RelNode树然后进行优化,最后进行执行。如下流程流转图:

【源码解析】flink sql执行源码概述:flink sql执行过程中有哪些阶段,这些阶段的源码大概位置在哪里,# flink源码,flink,sql,大数据
 

flink使用的是一款开源SQL解析工具Apache Calcite ,Calcite使用Java CC对sql语句进行了解析,转换为java/scala语言能够执行的逻辑。

 
Sql 的执行过程一般可以分为下图中的四个阶段,Calcite 同样也是这样:解析,校验,优化,执行:
【源码解析】flink sql执行源码概述:flink sql执行过程中有哪些阶段,这些阶段的源码大概位置在哪里,# flink源码,flink,sql,大数据

1. Sql语句解析成语法树阶段(SQL - > SqlNode)

对于flink中解析sql为SqlNode对象的流程为:

  • TableEnvironmentImpl是sql执行的入口类,TableEnvironmentImpl中提供了excuteSqlSqlQuery等方法用来执行DDL、DML等sql
  • sql执行时会先对sql进行解析,ParserImp是flink调用sql解析的实现类,ParserImpl#parse()方法中通过调用包装器对象CalciteParser#parse()方法创建并调用使用javacc生成的sql解析器
  • (FlinkSqlParserImpl)parseSqlStmtEof方法完成sql解析,并返回SqlNode对象。

具体calciteParser 的动作之后更新

【源码解析】flink sql执行源码概述:flink sql执行过程中有哪些阶段,这些阶段的源码大概位置在哪里,# flink源码,flink,sql,大数据

parse方法:负责将 SQL 查询字符串解析为抽象语法树(AST)

org.apache.flink.table.planner.delegation.ParserImpl
    /**
     * When parsing statement, it first uses {@link ExtendedParser} to parse statements. If {@link
     * ExtendedParser} fails to parse statement, it uses the {@link CalciteParser} to parse statements.
     * 先使用ExtendedParser进行解析如果解析失败了使用CalciteParser进行解析
     * @param statement input statement.
     * @return parsed operations.
     */
    @Override
    public List<Operation> parse(String statement) {
        //两种解析实例
        CalciteParser parser = calciteParserSupplier.get();
        FlinkPlannerImpl planner = validatorSupplier.get();
        //ExtendedParser解析
        Optional<Operation> command = EXTENDED_PARSER.parse(statement);
        if (command.isPresent()) {
            return Collections.singletonList(command.get());
        }
        //CalciteParser解析
        //解析为sqlNode
        SqlNodeList sqlNodeList = parser.parseSqlList(statement);
        List<SqlNode> parsed = sqlNodeList.getList();
        Preconditions.checkArgument(parsed.size() == 1, "only single statement supported");
        return Collections.singletonList(
                //解析为SOperation
                SqlToOperationConverter.convert(planner, catalogManager, parsed.get(0))
                        .orElseThrow(() -> new TableException("Unsupported query: " + statement)));
    }

将sql语句解析成sqlNode。
对应的表名、列名、with属性参数、主键、唯一键、分区键、水印、表注释、表操作(create table、alter table、drop table。。。)都放到SqlNode对象的对应属性中,SqlNode是一个树形结构也就是AST。
【源码解析】flink sql执行源码概述:flink sql执行过程中有哪些阶段,这些阶段的源码大概位置在哪里,# flink源码,flink,sql,大数据

 

2. SqlNode 验证(SqlNode – >Operation)

经过上面的第一步,会生成一个 SqlNode 对象,它是一个未经验证的抽象语法树,下面就进入了一个语法检查阶段,语法检查前需要知道元数据信息,这个检查会包括表名、字段名、函数名、数据类型的检查。

  • sql解析完成后执行sql校验,flink sql中增加了SqlNode转换为Operation的过程,sql校验是这个过程中完成。
  • 在SqlToOperationConvertver#convert()方法中完成这个转换,之后通过FlinkPlannerImpl#validate()方法对表、函数、字段等完成校验并基于生成的validated SqlNode生成对应的Operation。

【源码解析】flink sql执行源码概述:flink sql执行过程中有哪些阶段,这些阶段的源码大概位置在哪里,# flink源码,flink,sql,大数据

接着进入到SqlToOperationConverter.convert方法中

    /**
     * This is the main entrance for executing all kinds of DDL/DML {@code SqlNode}s, different
     * SqlNode will have it's implementation in the #convert(type) method whose 'type' argument is
     * subclass of {@code SqlNode}.
     * 转换DDL,DML(select比如?)sqlnode的主入口,不同的SqlNode有不同的convert实现。
     * 
     */
    public static Optional<Operation> convert(
            FlinkPlannerImpl flinkPlanner, CatalogManager catalogManager, SqlNode sqlNode) {
        //1.进行校验
        final SqlNode validated = flinkPlanner.validate(sqlNode);
        //2.转换为Operation
        return convertValidatedSqlNode(flinkPlanner, catalogManager, validated);
    }
 
   //将校验过的sqlnode转换为Operation
   private static Optional<Operation> convertValidatedSqlNode(
            FlinkPlannerImpl flinkPlanner, CatalogManager catalogManager, SqlNode validated) {
        SqlToOperationConverter converter =
                new SqlToOperationConverter(flinkPlanner, catalogManager);
	     。。。
	     else if (validated instanceof RichSqlInsert) {
	            return Optional.of(converter.convertSqlInsert((RichSqlInsert) validated));
	     } else if (validated.getKind().belongsTo(SqlKind.QUERY)) {
	            return Optional.of(converter.convertSqlQuery(validated));

但其实converter.convertSqlQuery包含了createSqlToRelConverter逻辑,即创建了SqlToRelConverter实例,用于转换RelNode。这里源码暂不展示。

 

3. 语义分析(Operation - > RelNode)

接着将 SqlNode 转换成 RelNode/RexNode,也就是生成相应的逻辑计划(Logical Plan),也就是最初版本的逻辑计划(Logical Plan)。

其中Operation中包含了RelNode的converter

源码见:org.apache.flink.table.planner.delegation.PlannerBase

  /** Converts a relational tree of [[ModifyOperation]] into a Calcite relational expression.
  将ModifyOperation转换为Calcite relational expression即RelNode
   */
  @VisibleForTesting
  private[flink] def translateToRel(modifyOperation: ModifyOperation): RelNode = {
    val dataTypeFactory = catalogManager.getDataTypeFactory
    modifyOperation match {
      case s: UnregisteredSinkModifyOperation[_] =>
        //relBuilder:
        val input = createRelBuilder.queryOperation(s.getChild).build()
        val sinkSchema = s.getSink.getTableSchema
        //校验relnode的 查询schema和sink schema是否一致,以及是否需要执行cast
        // validate query schema and sink schema, and apply cast if possible
        val query = validateSchemaAndApplyImplicitCast(
          input,
          catalogManager.getSchemaResolver.resolve(sinkSchema.toSchema),
          null,
          dataTypeFactory,
          getTypeFactory)
        LogicalLegacySink.create(
          query,
          s.getSink,
          "UnregisteredSink",
          ConnectorCatalogTable.sink(s.getSink, !isStreamingMode))
....

这里触发创建relnode的调用逻辑,这里在之后statementSet.execute()后执行。

//执行flink sql的调用
。。。
//解析sql -> sqlnode
StatementSet statementSet = SqlParser.parseSql(job, jarUrlList, tableEnv);
//sqlnode->relnode
TableResult execute = statementSet.execute();
	tableEnvironment.executeInternal(operations);
		TableEnvironmentImpl.translate
			PlannerBase.translate->translateToRel			

 

4. 优化阶段(RelNode - > optimize - >Transformation )

即对逻辑计划优化,根据前面生成的逻辑计划按照相应的规则进行优化。

接着看tableEnvironment.executeInternal中的translate方法

  override def translate(
      modifyOperations: util.List[ModifyOperation]): util.List[Transformation[_]] = {
    beforeTranslation()
    。。。
    //转换为relnode并放到一个map中
    val relNodes = modifyOperations.map(translateToRel)
    //优化逻辑计划
    val optimizedRelNodes = optimize(relNodes)
    //生成execGraph:执行图
    val execGraph = translateToExecNodeGraph(optimizedRelNodes, isCompiled = false)
    //生成transformations DAG
    val transformations = translateToPlan(execGraph)
    afterTranslation()
    transformations
  }

Calcite 的核心所在,优化器进行优化的地方,如过滤条件的下压(push down),在进行 join 操作前,先进行 filter 操作,这样的话就不需要在 join 时进行全量 join,减少参与 join 的数据量等。

 

5. 生成ExecutionPlan并执行

最终的执行计划转为Graph图,下面的流程与真正的java代码流程就一致了。

TableEnvironmentImpl.executeInternal中具体看executeInternal方法

TableEnvironmentImpl.executeInternal(
...
TableResultInternal result = executeInternal(transformations, sinkIdentifierNames);
...
)


    private TableResultInternal executeInternal(
            List<Transformation<?>> transformations, List<String> sinkIdentifierNames) {
         。。。
         //Translates the given transformations to a Pipeline.
         //将transformations转换为pipeline
        Pipeline pipeline =
                execEnv.createPipeline(
                        transformations, tableConfig.getConfiguration(), defaultJobName);
        try {
            // 执行pipeline
            //pipeline 其实就是StreamGraph
            JobClient jobClient = execEnv.executeAsync(pipeline);
         。。。
            }
		 。。。
		 //执行后返回结果
            return TableResult。。。

    }

通过生成的Transformation对象调用 execEnv.createPipeline,生成pipelinepipeline其实就StreamGraph便可以调用execEnv.executeAsync执行任务了。

 

二. 源码分析小结

上述描述了flink sql在内部执行过程进行的一些操作:
【源码解析】flink sql执行源码概述:flink sql执行过程中有哪些阶段,这些阶段的源码大概位置在哪里,# flink源码,flink,sql,大数据

这里我们从执行sql tEnv.executeSql(stmt)statementSet.addInsertSql(sql); parse、validate阶段)生成statementSet,然后执行statementSet.execute()optimize、Execute阶段) 触发任务执行。

 

parse、validate阶段

通过执行以下代码触发

StatementSet statementSet = SqlParser.parseSql(job, jarUrlList, tableEnv);

statementSet.addInsertSql(sql);
tEnv.executeSql(stmt)

通过调用 tEnv.executeSql(stmt)statementSet.addInsertSql(sql); 进行每个sql的解析,校验,具体:

  1. Sql语句解析成语法树阶段(SQL - > SqlNode)
  2. SqlNode 验证(SqlNode – >Operation),其中Operation中包含着RelNode的convert实例,为转换逻辑计划做提前准备

 
optimize、Execute阶段

接着执行如下代码

//sqlnode->relnode->优化->pipeline(StreamGraph)-> 执行并返回结果
TableResult execute = statementSet.execute();
//调用链:
	tableEnvironment.executeInternal(operations);
		TableEnvironmentImpl.translate
			PlannerBase.translate->translateToRel

经历如下几个过程:

sqlnode->relnode->优化->pipeline(StreamGraph)-> 执行并返回结果

 

这里我们主要看executeInternal的逻辑

    public TableResultInternal executeInternal(List<ModifyOperation> operations) {
        List<ModifyOperation> mapOperations = new ArrayList<>();
        for (ModifyOperation modify : operations) {
            //1.先执行CTAS sql语句, 并放到mapOperations中进行translate操作
            // execute CREATE TABLE first for CTAS statements
            if (modify instanceof CreateTableASOperation) {
                CreateTableASOperation ctasOperation = (CreateTableASOperation) modify;
                executeInternal(ctasOperation.getCreateTableOperation());
                mapOperations.add(ctasOperation.toSinkModifyOperation(catalogManager));
            } else {
               //2. 将其他非CTAS sqlnode放到mapOperations,进行translate操作
                mapOperations.add(modify);
            }
        }
        //translate主要的逻辑是:将所有的sqlNodes转换为relNodes,为初始的逻辑计划,然后优化逻辑计划,
        //接着翻译 ExecNodeGraph 为 Transformation DAG.
        List<Transformation<?>> transformations = translate(mapOperations);
        List<String> sinkIdentifierNames = extractSinkIdentifierNames(mapOperations);
        //  transformations转换为pipeline,最终执行pipeline即StreamGraph,然后返回结果
        TableResultInternal result = executeInternal(transformations, sinkIdentifierNames);
        if (tableConfig.get(TABLE_DML_SYNC)) {
            try {
                result.await();
            } catch (InterruptedException | ExecutionException e) {
                result.getJobClient().ifPresent(JobClient::cancel);
                throw new TableException("Fail to wait execution finish.", e);
            }
        }
        return result;
    }

 文章来源地址https://www.toymoban.com/news/detail-756882.html

到了这里,关于【源码解析】flink sql执行源码概述:flink sql执行过程中有哪些阶段,这些阶段的源码大概位置在哪里的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 源码解析Flink源节点数据读取是如何与checkpoint串行执行

    源码解析Flink源节点数据读取是如何与checkpoint串行执行 Flink版本:1.13.6 前置知识:源节点的Checkpoint是由Checkpointcoordinate触发,具体是通过RPC调用TaskManager中对应的Task的StreamTask类的performChecpoint方法执行Checkpoint。 本文思路:本文先分析checkpoint阶段,然后再分析数据读取阶段,

    2024年02月14日
    浏览(49)
  • 巡检过程中有哪些注意事项?智能巡检了解一下

    智能巡检系统是现场过程管理的生产力革命,由人工记录蝶化为掌上电脑运作,适用于设备运行值班记录、仓库/资产管理、设备巡检保养、安全巡更、机房值守、基站维护等一切重复性的工作管理。   安全巡检的目的在于识别信息系统存在的安全脆弱性、分析信息系统存在

    2024年02月07日
    浏览(46)
  • FlinkSQL-- sql-client及源码解析 -- flink-1.13.6

    本文基于flink-1.13.6 SQL Client: Init scripts and Statement Sets 这个版本极大地改进了 SQL 客户端的功能。现在 SQL Client 和 SQL 脚本都支持 通过Java 应用程序执行的几乎所有操作(从 TableEnvironment 以编程方式启动查询)。这意味着 SQL 用户在 SQL 部署中需要的代码少了很多。其中最核心的功能

    2023年04月27日
    浏览(38)
  • Java基础:单例模式,Spring源码中有哪些单例模式

    单例模式是一种常用的软件设计模式,其目的是确保一个类仅有一个实例,并提供一个全局访问点来获取这个唯一实例。在Java中,实现单例模式通常需要遵循以下几个关键原则: 私有化构造器 :将类的构造器声明为 private ,以防止外部代码通过 new 操作符直接创建该类的实

    2024年04月23日
    浏览(36)
  • 206.Flink(一):flink概述,flink集群搭建,flink中执行任务,单节点、yarn运行模式,三种部署模式的具体实现

    Flink官网地址:Apache Flink® — Stateful Computations over Data Streams | Apache Flink Flink是一个 框架 和 分布式处理引擎 ,用于对 无界 和 有界 数据流进行 有状态计算 。 无界流(流): 有定义流的开始,没有定义结束。会无休止产生数据 无界流数据必须持续处理 有界流(批): 有定

    2024年02月11日
    浏览(47)
  • Mybatis中执行Sql的执行过程

    MyBatis中执行SQL的过程可以分为以下几个步骤: 解析配置文件:在运行时,MyBatis会加载并解析配置文件(通常为 mybatis-config.xml ),获取数据库连接信息、映射文件等。 创建SqlSessionFactory:MyBatis使用SqlSessionFactory来创建SqlSession对象。SqlSessionFactory是一个线程安全的对象,应该

    2024年02月08日
    浏览(38)
  • SQL执行过程

    一条 select 语句的执行过程如上图所示 1、建立连接 连接器会校验你输入的用户名和密码是否正确,如果错误会返回提示,如果正确,连接器会查询当前用户对于的权限。连接器的作用就是校验用户权限 2、查询缓存 MySQL 中有个缓存的概念,当你在执行一条 SQL 查询语句时,

    2024年02月06日
    浏览(35)
  • 微信小程序的生命周期函数有哪些,以及执行过程?

    目录 1、应用的生命周期  2、页面的生命周期 3、组件的生命周期  4、执行顺序  小程序中,生命周期主要分成了三部分: ① 应用的生命周期        ② 页面的生命周期        ③ 组件的生命周期 1、应用的生命周期  在 app.js 里面调用,通过 App(Object) 函数用来注册一个小

    2024年02月08日
    浏览(46)
  • 【Git教程】(十七)发行版交付 — 概述及使用要求,执行过程及其实现,替代解决方案 ~

    对于每个项目或产品来说,发布版本的创建都需要一定的时间,其具体过程因各公司或组织的情况而异。 Git 无法用来指定项目进入发布阶段的时间。但我们可以利用标签和分支这两个强大的Git 工具来为发布进度设置一个很宽泛的时间区间。 在本章的工作流中,我们将以一个

    2024年04月28日
    浏览(42)
  • Flink源码-Task执行

    上一节我们分析到了Execution的生成,然后调用taskManagerGateway.submitTask方法提交task,提交的时候会将executionVertex封装成TaskDeploymentDescriptor,task的提交与执行涉及到了flink多个组件的配合,之前没有详细讲过,可能有的小伙伴有点不太清楚,这里我们花点时间介绍一下。 1.JobManager

    2024年02月03日
    浏览(36)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包