使用Calcite解析Sql做维表关联(二)

注册表

注册表包括源表、结果表。实时处理的数据源通常是kafka，针对不同的数据格式需要制定不同的反序列化方式，以json格式为例，如何将kafka的数据反序列化，将流转换为表，通常流的数据类型为Pojo、Tuple、Row等，为了能够通用化选择Row类型；结果表通常是 mysql 、hbase、es等，需要定义AppendStreamTableSink或者RetractStreamTableSink。

//以json格式为例
public class JsonDeserilization implements DeserializationSchema<Row> {

    private Map<String,String> fields; //fieldName->fieldType
    private RowTypeInfo rowTypeInfo;
    private TypeInformation<?>[] typeInformations;
    private String[] fieldNames;
    //传入的参数解析create语句得到
    public JsonDeserilization(String[] fieldNames,TypeInformation<?>[] typeInformations){
        this.fieldNames=fieldNames;
        this.typeInformations=typeInformations;
        this.rowTypeInfo=new RowTypeInfo(typeInformations,fieldNames);
    }

    @Override public Row deserialize(byte[] message) throws IOException {
        String msg=new String(message);
        Row row=new Row(fieldNames.length);
        JSONObject jsonObject=JSONObject.parseObject(msg);
        for(int i=0;i<fieldNames.length;i++){
            if(typeInformations[i].getTypeClass()==String.class){
                row.setField(i,jsonObject.getString(fieldNames[i]));
            }
            if(typeInformations[i].getTypeClass()==Integer.class){
                row.setField(i,jsonObject.getInteger(fieldNames[i]));
            }
        }
        return row;
    }
......
}

注册表：

//kafka json 类型
public static void registerSourceTable(TableInfo tableInfo,StreamTableEnvironment tblEnv){

        Properties props=tableInfo.getProps();
        String tableName=tableInfo.getTableName();
        StreamExecutionEnvironment env=tblEnv.execEnv();
        if("kafka".equals(props.getProperty("type"))) {
            Properties kafkaPros = new Properties();
            props.forEach((k, v) -> {
                if (k.toString().startsWith("kafka.")) {
                    kafkaPros.setProperty(k.toString().replace("kafka.", ""), v.toString());
                }
            });
            String topic = props.getProperty("kafka.topic");
            FlinkKafkaConsumer<Row> consumer011 = new FlinkKafkaConsumer<Row>(topic,
                    new JsonDeserilization(tableInfo.getFieldNames(), tableInfo.getFieldTypes()), kafkaPros);
            DataStream<Row> ds = env.addSource(consumer011);
            tblEnv.registerDataStreamInternal(tableName, ds);
        }
    }

public static void registerSinkTable(TableInfo tableInfo,StreamTableEnvironment tblEnv){
        Properties props=tableInfo.getProps();
        String tableName=tableInfo.getTableName();
        if("console".equals(props.getProperty("type"))){
            ConsoleTableSink consoleTableSink=new ConsoleTableSink();
            tblEnv.registerTableSink(tableName,tableInfo.getFieldNames(),tableInfo.getFieldTypes(),consoleTableSink);
        }
    }

ConsoleTableSink 实现了RetractStreamTableSink <Row> ，直接将数据原样输出到控制台。

Join实现

得到解析后的SqlJoin节点，获取源表、维表信息，首先将源表转换为流：

    SqlJoin sqlJoin=(SqlJoin)sqlNode1;

    String leftTableName=parseTableName(sqlJoin.getLeft()); //表名称
    String rightTableName=parseTableName(sqlJoin.getRight());
    
    TableInfo leftTableInfo=tableInfoMap.get(leftTableName);//表信息
    TableInfo rightTable=tableInfoMap.get(rightTableName);

    String leftAlias=paserAliasTableName(sqlJoin.getLeft()); //别名
    String rightAlias=paserAliasTableName(sqlJoin.getRight());

    Table leftTable=tblEnv.sqlQuery("select * from " + leftTableName);
    DataStream leftStream = tblEnv.toAppendStream(leftTable,Row.class); //转换后的流

接下来将流表与维表进行关联查询，根据维表根据设置的不同属性：同步/异步查询、cache/nocache方式、查询不同的外部存储等，需要实现不同的查询方式。以异步查询mysql为例分析：需要根据维表定义的字段、join的关联条件解析生成一条sql语句，根据流入数据解析出sql的查询条件值，然后查询得到对应的维表值，将流入数据与查询得到的维表数据拼接起来输出到下游：

public class MySqlAsyncFunction extends RichAsyncFunction<Row,Row> {

    private Connection connection;
    private String sqlTemplate;

    private String url;
    private String username;
    private String password;
    private String tableName;

    private int idx; //条件值在流入数据的位置
    private int inLength; //流入字段数
    private int outLength; //输出字段数
    private int sideLength; //维表查询字段数

    public MySqlAsyncFunction(SqlJoin sqlJoin,TableInfo sideTableInfo,TableInfo leftTableInfo){

        Properties props=sideTableInfo.getProps();
        this.url=props.getProperty("url");
        this.username=props.getProperty("username");
        this.password=props.getProperty("password");
        this.tableName=props.getProperty("tableName");

        String rightField=parseCondition(sqlJoin,false);
        genSqlTemplate(sideTableInfo,rightField);


        String leftField=parseCondition(sqlJoin,true);
        for (int i = leftTableInfo.getFieldNames().length - 1; i >= 0; i--) {
            if(leftField.equals(leftTableInfo.getFieldNames()[i])){
                this.idx=i;
                break;
            }
        }
        inLength=leftTableInfo.getFieldNames().length;
        sideLength=sideTableInfo.getFieldNames().length;
        outLength=inLength+sideLength;
    }

    @Override public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        this.connection= DriverManager.getConnection(url,username,password);
    }
              //这里还是一个同步查询，没有使用异步方式，需要使用一部mysql客户端
    @Override public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Exception {

        String v=(String)input.getField(idx); //获取条件值
        PreparedStatement preparedStatement=connection.prepareStatement(sqlTemplate);
        preparedStatement.setString(1,v);
        ResultSet rs=preparedStatement.executeQuery();
        boolean isJoin=false;
        while (rs.next()){
               isJoin=true;
               Row row=new Row(outLength);
               for(int i=0;i<input.getArity();i++){
                   row.setField(i,input.getField(i));
               }
               //直接将维表数据补齐在流数据后面
               for(int i=0;i<sideLength;i++){
                   row.setField(inLength+i,rs.getObject(i+1));
               }
               resultFuture.complete(Collections.singletonList(row));
           }
        if(!isJoin) resultFuture.complete(null);
    }
    //解析on 条件的左右表字段名称， 这里只解析了一个关联条件
    private String parseCondition(SqlJoin sqlJoin,boolean isLeft){
        SqlNode condition=sqlJoin.getCondition();
        SqlBasicCall sqlBasicCall=(SqlBasicCall)condition;
        String name=SqlExec.paserAliasTableName(isLeft?sqlJoin.getLeft():sqlJoin.getRight());
        SqlIdentifier sqlIdentifier1=(SqlIdentifier)sqlBasicCall.operands[0];
        if(name.equals(sqlIdentifier1.names.get(0))){
            return sqlIdentifier1.names.get(1);
        }

        SqlIdentifier sqlIdentifier2=(SqlIdentifier)sqlBasicCall.operands[1];
        if(name.equals(sqlIdentifier2.names.get(0))){
            return sqlIdentifier2.names.get(1);
        }
        return null;
    }

   //查询sql
    private void genSqlTemplate(TableInfo tableInfo,String condition){
        StringBuilder sql=new StringBuilder();
        StringBuilder selects=new StringBuilder();
        sql.append("select ");
        for(String field : tableInfo.getFieldNames()){
            selects.append(field);
            selects.append(",");
        }
        sql.append(selects.substring(0,selects.lastIndexOf(",")));
        sql.append(" from ").append(this.tableName);
        if(condition!=null)
            sql.append(" where ").append(condition).append("=?");
        this.sqlTemplate=sql.toString();
    }
}

到目前为止获取了join之后的结果数据，但是有一点需要考虑，流表定义字段名称与维表定义的字段名称可能会相同，那么在将流转换为表时就存在相同的字段，因此需要对相同的字段重命名：

TableInfo leftTableInfo=tableInfoMap.get(leftTableName);
          TableInfo rightTable=tableInfoMap.get(rightTableName);

          List<String> newFields=new ArrayList<>(); //join之后流的字段名称
          List<TypeInformation> newTypes=new ArrayList<>(); //join之后流的数据类型

        //需要做字段解析 原始表名-原始字段名称-新字段名称
                HashBasedTable hashBasedTable=HashBasedTable.create();
                int i=0;
                for(String field:leftTable.getSchema().getFieldNames()){
                    hashBasedTable.put(leftAlias,field,field);
                    newFields.add(field);
                    newTypes.add(leftTable.getSchema().getFieldType(i).get());
                    i++;
                }
                i=0;
                for(String field:rightTable.getFieldNames()){
                    String newField=field;
                    if(hashBasedTable.containsColumn(field)){
                        newField=field+"0";
                    }
                    hashBasedTable.put(rightAlias,field,newField);
                    newFields.add(newField);
                    newTypes.add(rightTable.getFieldTypes()[i]);
                    i++;
                }

                String newTableNameAlias=leftAlias+"_"+rightAlias;
                String newTableName=leftTableName+"_"+rightTableName;
                hashBasedTableMap.put(newTableNameAlias,hashBasedTable);
               //outType 表示关联之后的流数据类型
                RowTypeInfo outType=new RowTypeInfo(newTypes.toArray(new TypeInformation[]{}),newFields.toArray(new String[]{}));
                DataStream dsOut=AsyncDataStream.unorderedWait(leftStream,new MySqlAsyncFunction(sqlJoin,rightTable,leftTableInfo),10,
                        TimeUnit.SECONDS);
                dsOut.getTransformation().setOutputType(outType);
                tblEnv.registerDataStream(newTableName,dsOut); //将join之后的流注册成为表

同样也需要对insert部分的select字段根据上面得到新的字段名称hashBasedTable进行替换：

  SqlInsert sqlInsert=(SqlInsert)sqlNode1;
  SqlSelect source=(SqlSelect)sqlInsert.getSource();

 SqlBasicCall sqlBasicCall=(SqlBasicCall) source.getFrom();
 String newAlias=sqlBasicCall.operands[1].toString(); //新表的别名
  HashBasedTable hashBasedTable=hashBasedTableMap.get(newAlias);

    SqlNodeList sqlNodeList=source.getSelectList();
    int i=0;
  for(SqlNode x: sqlNodeList.getList()){
    SqlIdentifier sqlIdentifier=(SqlIdentifier)x;
    String tableAlias=sqlIdentifier.names.get(0);
    String field=sqlIdentifier.names.get(1);    String newFieldName=hashBasedTable.get(tableAlias,field).toString();
    sqlIdentifier=sqlIdentifier.setName(0,newAlias);//替换为新的表别名称
    sqlIdentifier=sqlIdentifier.setName(1,newFieldName);//替换为新的字段名称
    sqlNodeList.set(i,sqlIdentifier); 
    i++;
 }

 tblEnv.sqlUpdate(sqlInsert.toString()); //执行insert 语句

总结

以上提供了流表join维表的sql实现思路以及部分demo代码的参考，但是其远远达不到工程上的要求，在实际使用中需要要考虑更多的因素：复杂嵌套的sql、时间语义支持、自定义函数支持等。推荐一个开源项flinkStreamSql, 地址为： https://github.com/DTStack/flinkStreamSQL ， 丰富的语义支持、不同类型的源插件支持等。

关注回复 Flink

获取更多系列