flink 多表join的例子

今天寫了一個稍微復雜的例子， 實現了類似mysql group_concat 功能，記錄一下
MapToString 參考bug 那篇博客

public static void main(String[] arg) throws Exception {

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tableEnv = new BatchTableEnvironment(env, TableConfig.DEFAULT());
        tableEnv.registerFunction("mapToString", new MapToString());

        getProjectInfo(env,tableEnv);
        getProject(env,tableEnv);
        joinTableProjectWithInfo(tableEnv);

        Table query = tableEnv.sqlQuery("select id, name, type from result_agg");
        DataSet<Row> ds=  tableEnv.toDataSet(query, Row.class);
        ds.print();
        ds.writeAsText("/home/test", WriteMode.OVERWRITE);
        env.execute("multiple-table");          
    }

    public static void getProjectInfo(ExecutionEnvironment env,BatchTableEnvironment tableEnv) {

        TypeInformation[] fieldTypes = new TypeInformation[] { BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO };
        String[] fieldNames = new String[] { "id",  "type" };
        RowTypeInfo rowTypeInfo = new RowTypeInfo(fieldTypes, fieldNames);
        JDBCInputFormat jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat().setDrivername("com.mysql.jdbc.Driver")
                .setDBUrl("jdbc:mysql://ip:3306/space?characterEncoding=utf8")
                .setUsername("user").setPassword("pwd")
                .setQuery("select project_fid, cast(project_info_type as CHAR) as type from project").setRowTypeInfo(rowTypeInfo).finish();
        DataSource<Row> s = env.createInput(jdbcInputFormat);   
        tableEnv.registerDataSet("project_info", s);
        aggProjectInfo(tableEnv,"project_info");
    }

    public static void aggProjectInfo(BatchTableEnvironment tableEnv, String tableName) {   
        Table tapiResult = tableEnv.scan(tableName);
        tapiResult.printSchema();               
        Table query = tableEnv.sqlQuery("select id, mapToString(collect(type)) as type from project_info group by id");
        tableEnv.registerTable(tableName+"_agg", query);        
        tapiResult = tableEnv.scan(tableName+"_agg");
        tapiResult.printSchema();   
    }

    public static void getProject(ExecutionEnvironment env,BatchTableEnvironment tableEnv) {

        TypeInformation[] fieldTypes = new TypeInformation[] { BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO };
        String[] fieldNames = new String[] { "pid",  "name" };
        RowTypeInfo rowTypeInfo = new RowTypeInfo(fieldTypes, fieldNames);
        JDBCInputFormat jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat().setDrivername("com.mysql.jdbc.Driver")
                .setDBUrl("jdbc:mysql://ip:3306/space?characterEncoding=utf8")
                .setUsername("user").setPassword("pwd")
                .setQuery("select fid, project_name  from t_project").setRowTypeInfo(rowTypeInfo).finish();
        DataSource<Row> s = env.createInput(jdbcInputFormat);
        tableEnv.registerDataSet("project", s);

    }

    public static void joinTableProjectWithInfo(BatchTableEnvironment tableEnv) {
        Table result =tableEnv.sqlQuery("select a.pid as id , a.name , b.type  from project a inner join  project_info_agg  b on a.pid=b.id");
        tableEnv.registerTable("result_agg", result);
        result.printSchema();
    }