Shulou(Shulou.com)06/03 Report--

Today I wrote a slightly more complex example that implements a function similar to mysql group_concat. Record it.

MapToString refers to the blog post of 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 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 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 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 an inner join project_info_agg b on a.pid=b.id"); tableEnv.registerTable ("result_agg", result); result.printSchema ();}

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.