Dear All, 近期在research Hive on Flink的一些特性,之前有一个Mapreduce的任务,语句【HQL】如下:
逻辑简单,全连接两个表(table1和table2),将结果写进新表table3。table1、table2和table3三张表结构相同,都有35个字段,join的时候用id作为连接,进行比较,如果table2的字段值不为null,或者不为空,就用table2的字段,反之用table1字段。最后把结果写进新表table3。 到目前为止,已经执行了17h,还没有结束,看数据流,好像快完了,不知道我的使用方法是否合理? 【HQL 】 insert into table3 select if(t2.id is not null and t2.id <> '', t2.id, t1.id) as id ,if(t2.field2 is not null and t2.field2 <> '', t2.field2, t1.field2) as field2 ...... ...... ...... ...... ...... ...... ...... ...... ,if(t2.field35 is not null and field35.dt <> '', field35.dt , field35.dt ) as field35 from ( select * from table1 where (id is not null and id <> '') ) as t1 full join ( select * from table2 where (id is not null and id <> '') ) as t2 on (t1.id = t2.id) 代码如下: public class FlinkHiveIntegration1 { public static void main(String[] args) throws Exception { EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().build(); TableEnvironment tableEnv = TableEnvironment.create(settings); String name = "myhive"; String database = "mydatabase"; String version = "1.1.0-cdh5.8.3"; HiveConf hiveConf = new HiveConf(); hiveConf.setVar(HiveConf.ConfVars.METASTOREURIS,"thrift://*******:9083"); hiveConf.setVar(HiveConf.ConfVars.METASTOREWAREHOUSE,"hdfs://nameservice1/user/hive/warehouse"); HiveCatalog hive = new HiveCatalogTest(name, database, hiveConf , version); tableEnv.registerCatalog(name, hive); tableEnv.useCatalog(name); tableEnv.useDatabase(database); String HQL = HQL ; tableEnv.getConfig().addConfiguration(new Configuration().set(CoreOptions.DEFAULT_PARALLELISM, 8)); tableEnv.executeSql(HQL ); } } <http://apache-flink.147419.n8.nabble.com/file/t1162/hiveonFlink.png> ----- Thanks! Jacob -- Sent from: http://apache-flink.147419.n8.nabble.com/