答复: Flink 1.11里如何parse出未解析的执行计划

Thu, 22 Oct 2020 23:27:49 -0700 

Hi,


我的做法如我所说，是用反射将parser拿出来的，比较hack但是很简单而且很稳妥


代码差不多就是下面这个样子


Flink version： custom version base on 1.11.x


@PostConstruct

private void setup() throws NoSuchFieldException, IllegalAccessException {
        final StreamTableEnvironmentImpl env = (StreamTableEnvironmentImpl) 
support.getStreamTableEnvironment();
        final Field field = 
env.getParser().getClass().getDeclaredField("calciteParserSupplier");
        field.setAccessible(true);
        // 普通的parser
        final Supplier<CalciteParser> defaultSupplier = 
(Supplier<CalciteParser>) field.get(env.getParser());
        this.defaultSupplier = defaultSupplier;
        env.getConfig().setSqlDialect(SqlDialect.HIVE);
        final Field field2 = 
env.getParser().getClass().getDeclaredField("calciteParserSupplier");
        field2.setAccessible(true);
        // hive的parser
        final Supplier<CalciteParser> hiveSupplier = (Supplier<CalciteParser>) 
field.get(env.getParser());
        this.hiveSupplier = hiveSupplier;
    }

WARN:
这种做法带来的比较不好的点是，项目的依赖会比较多。我这边恰好是一个在一个单独的且包含Flink依赖的服务中做的这件事从而规避了这个问题。
如果要参考上面这种做法的话，注意评估依赖的问题啦~




________________________________
发件人: silence <slle...@aliyun.com.INVALID>
发送时间: 2020年10月22日 12:05:38
收件人: user-zh@flink.apache.org
主题: Re: Flink 1.11里如何parse出未解析的执行计划

我简单写了一下仅供参考

import org.apache.calcite.config.Lex;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlNodeList;
import org.apache.calcite.sql.parser.SqlParseException;
import org.apache.calcite.sql.parser.SqlParser;
import org.apache.flink.sql.parser.impl.FlinkSqlParserImpl;
import org.apache.flink.sql.parser.validate.FlinkSqlConformance;

/**
 * @author: silence
 * @date: 2020/10/22
 */
public class Test {
    public static void main(String[] args) throws SqlParseException {
        String sql = "xxx";
        SqlParser.Config sqlParserConfig = SqlParser
                .configBuilder()
                .setParserFactory(FlinkSqlParserImpl.FACTORY)
                .setConformance(FlinkSqlConformance.DEFAULT)
                .setLex(Lex.JAVA)
                .setIdentifierMaxLength(256)
                .build();
        SqlParser sqlParser = SqlParser.create(sql, sqlParserConfig);
        SqlNodeList sqlNodes = sqlParser.parseStmtList();
        for (SqlNode sqlNode : sqlNodes) {
            //do something
        }
    }
}



--
Sent from: http://apache-flink.147419.n8.nabble.com/

回复