个推基于Flink SQL建设实时数仓实践

/opt/flink/bin/flink run -m yarn-cluster -ynm KafkaSourceHbaseSinkCaseTestSql \-c ${mainClassName} \${jarPath} \--flink.parallelism 40 \--mode stream \--sql.file.path ${sqlFile}

create table kafka_table(  ts  bigint,  username string,  num bigint) with (  'connector' = 'kafka',  'topic' = 'test',  'properties.bootstrap.servers' = '',  'properties.group.id' = 'test-consumer001',  'scan.startup.mode' = 'latest-offset',  'format' = 'csv')
create table sink_table(  ts  bigint,  username string,  num bigint) with (  'connector' = 'kafka',  'topic' = 'test2',  'properties.bootstrap.servers' = '',  'format' = 'json')
insert into sink_table select * from kafka_table

* {{{-*  val sourceTable = tEnv.scan("test_table").select('a, 'b, 'c)*  val leftTable = sourceTable.filter('a > 0).select('a as 'a1, 'b as 'b1)*  val rightTable = sourceTable.filter('c.isNotNull).select('b as 'b2, 'c as 'c2)*  val joinTable = leftTable.join(rightTable, 'a1 === 'b2)*  joinTable.where('a1 >= 70).select('a1, 'b1).writeToSink(sink1)*  joinTable.where('a1 < 70 ).select('a1, 'c2).writeToSink(sink2)* }}}** the RelNode DAG is:** {{{-* Sink(sink1)     Sink(sink2)*    |               |* Project(a1,b1)  Project(a1,c2)*    |               |* Filter(a1>=70)  Filter(a1<70)*       \          /*        Join(a1=b2)*       /           \* Project(a1,b1)  Project(b2,c2)*      |             |* Filter(a>0)     Filter(c is not null)*      \           /*      Project(a,b,c)*          |*       TableScan* }}}* This [[RelNode]] DAG will be decomposed into three [[RelNodeBlock]]s, the break-point* is the [[RelNode]](`Join(a1=b2)`) which data outputs to multiple [[LegacySink]]s.* <p>Notes: Although `Project(a,b,c)` has two parents (outputs),* they eventually merged at `Join(a1=b2)`. So `Project(a,b,c)` is not a break-point.* <p>the first [[RelNodeBlock]] includes TableScan, Project(a,b,c), Filter(a>0),* Filter(c is not null), Project(a1,b1), Project(b2,c2) and Join(a1=b2)* <p>the second one includes Filter(a1>=70), Project(a1,b1) and Sink(sink1)* <p>the third one includes Filter(a1<70), Project(a1,c2) and Sink(sink2)* <p>And the first [[RelNodeBlock]] is the child of another two.** The [[RelNodeBlock]] plan is:* {{{-* RelNodeBlock2  RelNodeBlock3*        \            /*        RelNodeBlock1* }}}

create table source_table(  data string,  topic string,) with ( 'connector' = 'kafka', 'topic' = 'topic', 'properties.bootstrap.servers' = '', 'format' = 'csv')
create table sink_table1(  data string,) with ( 'connector' = 'kafka', 'topic' = 'topic2', 'properties.bootstrap.servers' = '', 'format' = 'csv')
create table sink_table2(  data string,) with ( 'connector' = 'kafka', 'topic' = 'topic2', 'properties.bootstrap.servers' = '', 'format' = 'csv')
insert into sink_table1 select * from (select SBSTR(data, 0, 6) data,topic from source_table) where topic='topic1'
insert into sink_table2 select * from (select SBSTR(data, 0, 6) data,topic from source_table) where topic='topic2'

/**    * Reuse common sub-plan in different RelNode tree, generate a RelNode dag    *    * @param relNodes RelNode trees    * @return RelNode dag which reuse common subPlan in each tree    */  private def reuseRelNodes(relNodes: Seq[RelNode], tableConfig: TableConfig): Seq[RelNode] = {    val findOpBlockWithDigest = tableConfig.getConfiguration.getBoolean(      RelNodeBlockPlanBuilder.TABLE_OPTIMIZER_REUSE_OPTIMIZE_BLOCK_WITH_DIGEST_ENABLED)    if (!findOpBlockWithDigest) {      return relNodes    }
    // reuse sub-plan with same digest in input RelNode trees.    val context = new SubplanReuseContext(true, relNodes: _*)    val reuseShuttle = new SubplanReuseShuttle(context)    relNodes.map(_.accept(reuseShuttle))  }

// 创建视图StreamTableEnvironment.executeSql("create view as select SBSTR(data, 0, 6) data,topic from source_table);
// 注册表Table table = StreamTableEnvironment.executeSql("select SBSTR(data, 0, 6) data,topic from source_table");
StreamTableEnvironment.registerTable("tmp", table);

REGISTER TABLE TABLE_NAME AS SELECT_QUERY
// 示例register table tmp as select SBSTR(data, 0, 6) data,topic from source_table

package org.apache.flink.sql.parser.ddl;
import org.apache.calcite.sql.*;import org.apache.calcite.sql.parser.SqlParserPos;import javax.annotation.Nonnull;
import java.util.Collections;import java.util.List;
public class SqlRegisterTable extends SqlCall {public static final SqlSpecialOperator OPERATOR = new SqlSpecialOperator("REGISTER TABLE", SqlKind.OTHER_DDL);
private SqlIdentifier tableName;private SqlNode query;     // 部分代码省略 }

SqlRegisterTable SqlRegisterTable() :  {    SqlIdentifier tableName = null;    SqlNode query = null;    SqlParserPos pos;    }  {    <REGISTER> <TABLE> { pos = getPos();}    tableName = CompoundIdentifier()    <AS>    query = OrderedQueryOrExpr(ExprContext.ACCEPT_QUERY)    {        return new SqlRegisterTable(pos, tableName, query);        }  }

package org.apache.flink.table.operations.ddl;
import org.apache.flink.table.catalog.ObjectIdentifier;import org.apache.flink.table.operations.Operation;import org.apache.flink.table.operations.QueryOperation;
public class RegisterTableOperation implements Operation{  private final ObjectIdentifier tableIdentifier;  private final QueryOperation query;
    // 省略部分代码}

private Operation convertRegisterTable(SqlRegisterTable sqlRegisterTable) {    UnresolvedIdentifier unresolvedIdentifier = UnresolvedIdentifier.of(sqlRegisterTable.fullTableName());    ObjectIdentifier identifier = catalogManager.qualifyIdentifier(unresolvedIdentifier);    PlannerQueryOperation operation = toQueryOperation(flinkPlanner, validateQuery);
    return new RegisterTableOperation(identifier, operation);  }

@Override  public TableResult executeSql(String statement) {    List<Operation> operations = parser.parse(statement);
    if (operations.size() != 1) {      throw new TableException(UNSUPPORTED_QUERY_IN_EXECUTE_SQL_MSG);    }
    return executeOperation(operations.get(0));  }