使用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

获取更多系列