Hi, 我看其中一个 condition 是 `t1.uid = t2.refer_id` 其中 uid 是 bigint 类型，refer_id 是 varchar 类型。 你再确认下？
Best, Hailong Wang At 2020-10-20 08:55:34, "Dream-底限" <[email protected]> wrote: >hi、 >我查看了一下，join条件类型是一样的，我这面下游sink使用jdbc时候是可以运行的，但是转换为datastream时候失败了，下面是程序及异常： > >streamTableEnv.executeSql(kafkaDDL);//ddl语句见下面日志 > > >Table table = streamTableEnv.sqlQuery("SELECT cast(t1.id as bigint) as >register_id,cast(t1.uid as bigint) as asi_uid,cast(null as bigint) as >person_uuid,cast(t1.app_id as bigint) as app_id,cast(t1.country_id as >bigint) as country_id,t2.channel_id as channel_id,t2.device_id as >device_id,t2.adjust_id as adjust_id,t2.google_adid as >google_adid,cast(t3.referrer as bigint) as referrer,t3.login_pwd as >login_pwd,cast(t1.sync_data_flag as int) as >sync_data_flag,t3.phone_number as >register_phone_number,cast(t2.device_type as int) as >device_type,t2.imei as imei,t2.device_model as >device_model,t2.os_version as os_version,t2.app_name as >app_name,t2.app_version as app_version,t2.app_package_name as >app_package_name,cast(t2.network_type as string) as >network_type,t2.wifi_mac as wifi_mac,t2.lgt as longitude,t2.lat as >latitude,cast(null as string) as geo_hash7,t2.ip as >ip,unix_timestamp(t1.create_time,'yyyy-MM-dd HH:mm:ss') as >register_time,UNIX_TIMESTAMP() as etl_time from (SELECT >`uid`,`update_time`,`adjust_id`,`create_time`,`source_type`,`sync_data_flag`,`id`,`app_id`,`country_id` >FROM (SELECT >`rowData`.`uid`,`rowData`.`update_time`,`rowData`.`adjust_id`,`rowData`.`create_time`,`rowData`.`source_type`,`rowData`.`sync_data_flag`,`rowData`.`id`,`rowData`.`app_id`,`rowData`.`country_id`,`binlogTime`,ROW_NUMBER() >OVER (PARTITION BY `rowData`.`uid`,`rowData`.`id` ORDER BY >`binlogTime` desc) AS rownum FROM >asi_user_user_service_t_user_register_source) WHERE rownum = 1) t1 >left join (SELECT >`refer_id`,`device_id`,`channel_id`,`device_type`,`imei`,`adjust_id`,`google_adid`,`device_model`,`os_version`,`app_name`,`app_package_name`,`app_version`,`ip`,`network_type`,`wifi_mac`,`lgt`,`lat`,`event_id`,`country_id`,`uid`,`create_time`,`api_version` >FROM (SELECT >`data`.`data`.`refer_id`,`data`.`data`.`device_id`,`data`.`data`.`channel_id`,`data`.`data`.`device_type`,`data`.`data`.`imei`,`data`.`data`.`adjust_id`,`data`.`data`.`google_adid`,`data`.`data`.`device_model`,`data`.`data`.`os_version`,`data`.`data`.`app_name`,`data`.`data`.`app_package_name`,`data`.`data`.`app_version`,`data`.`data`.`ip`,`data`.`data`.`network_type`,`data`.`data`.`wifi_mac`,`data`.`data`.`lgt`,`data`.`data`.`lat`,`data`.`event_id`,`data`.`country_id`,`data`.`uid`,`data`.`create_time`,`data`.`api_version`,ROW_NUMBER() >OVER (PARTITION BY `data`.`data`.`refer_id`,`data`.`event_id` ORDER BY >createTime desc) AS rownum FROM eventDeviceInfo where >`data`.`event_id`=1002) WHERE rownum = 1) t2 on t1.uid = t2.refer_id >left join (SELECT >`register_channel_source`,`last_login_time`,`create_time`,`language`,`avatar`,`login_pwd`,`email_status`,`storage_source`,`uid`,`referrer`,`update_time`,`nickname`,`phone_number`,`sync_data_flag`,`id`,`country_id`,`email`,`status` >FROM (SELECT >`rowData`.`register_channel_source`,`rowData`.`last_login_time`,`rowData`.`create_time`,`rowData`.`language`,`rowData`.`avatar`,`rowData`.`login_pwd`,`rowData`.`email_status`,`rowData`.`storage_source`,`rowData`.`uid`,`rowData`.`referrer`,`rowData`.`update_time`,`rowData`.`nickname`,`rowData`.`phone_number`,`rowData`.`sync_data_flag`,`rowData`.`id`,`rowData`.`country_id`,`rowData`.`email`,`rowData`.`status`,`binlogTime`,ROW_NUMBER() >OVER (PARTITION BY `rowData`.`uid` ORDER BY `binlogTime` desc) AS >rownum FROM asi_user_user_service_t_user) WHERE rownum = 1) t3 on >t1.uid=t3.uid"); > >table.printSchema(); >streamTableEnv.toRetractStream(table, >Types.ROW(TypeConversions.fromDataTypeToLegacyInfo(table.getSchema().getFieldDataTypes()))).print(); >streamExecEnv.execute("kafka-json-test"); > > > > >CREATE TABLE eventDeviceInfo (`data` ROW<`event_id` >BIGINT,`country_id` BIGINT,`uid` BIGINT,`create_time` BIGINT,`data` >ROW<`refer_id` STRING,`device_id` STRING,`channel_id` >STRING,`device_type` BIGINT,`imei` STRING,`adjust_id` >STRING,`google_adid` STRING,`device_model` STRING,`os_version` >STRING,`app_name` STRING,`app_package_name` STRING,`app_version` >STRING,`ip` STRING,`network_type` BIGINT,`wifi_mac` STRING,`lgt` >DECIMAL(38,18),`lat` DECIMAL(38,18)>,`api_version` >STRING>,`createTime` BIGINT) WITH ('connector' = 'kafka-0.11','topic' >= 'eventDeviceInfo','properties.bootstrap.servers' = >'127.0.0.1:9092','properties.group.id' = >'test','properties.max.poll.records' = >'1000','properties.flink.poll-timeout' = '10','scan.startup.mode' = >'earliest-offset','format' = 'json','json.fail-on-missing-field' = >'false','json.ignore-parse-errors' = 'true') >CREATE TABLE asi_user_user_service_t_user (`binlogTime` >BIGINT,`rowData` ROW<`register_channel_source` >STRING,`last_login_time` STRING,`create_time` STRING,`language` >STRING,`avatar` STRING,`login_pwd` STRING,`email_status` >STRING,`storage_source` STRING,`uid` STRING,`referrer` >STRING,`update_time` STRING,`nickname` STRING,`phone_number` >STRING,`sync_data_flag` STRING,`id` STRING,`country_id` STRING,`email` >STRING,`status` STRING>) WITH ('connector' = 'kafka-0.11','topic' = >'asi_user_user_service_t_user','properties.bootstrap.servers' = >'127.0.0.1:9092','properties.group.id' = >'test','properties.max.poll.records' = >'1000','properties.flink.poll-timeout' = '10','scan.startup.mode' = >'earliest-offset','format' = 'json','json.fail-on-missing-field' = >'false','json.ignore-parse-errors' = 'true') >CREATE TABLE asi_user_user_service_t_user_register_source >(`binlogTime` BIGINT,`rowData` ROW<`uid` BIGINT,`update_time` >STRING,`adjust_id` STRING,`create_time` STRING,`source_type` >STRING,`sync_data_flag` STRING,`id` STRING,`app_id` >STRING,`country_id` BIGINT>) WITH ('connector' = 'kafka-0.11','topic' >= 'asi_user_user_service_t_user_register_source','properties.bootstrap.servers' >= '127.0.0.1:9092','properties.group.id' = >'test','properties.max.poll.records' = >'1000','properties.flink.poll-timeout' = '10','scan.startup.mode' = >'earliest-offset','format' = 'json','json.fail-on-missing-field' = >'false','json.ignore-parse-errors' = 'true') >root > |-- register_id: BIGINT > |-- asi_uid: BIGINT > |-- person_uuid: BIGINT > |-- app_id: BIGINT > |-- country_id: BIGINT > |-- channel_id: STRING > |-- device_id: STRING > |-- adjust_id: STRING > |-- google_adid: STRING > |-- referrer: BIGINT > |-- login_pwd: STRING > |-- sync_data_flag: INT > |-- register_phone_number: STRING > |-- device_type: INT > |-- imei: STRING > |-- device_model: STRING > |-- os_version: STRING > |-- app_name: STRING > |-- app_version: STRING > |-- app_package_name: STRING > |-- network_type: STRING > |-- wifi_mac: STRING > |-- longitude: DECIMAL(38, 18) > |-- latitude: DECIMAL(38, 18) > |-- geo_hash7: STRING > |-- ip: STRING > |-- register_time: BIGINT > |-- etl_time: BIGINT NOT NULL > > >org.apache.flink.table.api.TableException: BIGINT and >VARCHAR(2147483647) does not have common type now > > at > org.apache.flink.table.planner.plan.rules.logical.JoinConditionTypeCoerceRule$$anonfun$onMatch$1.apply(JoinConditionTypeCoerceRule.scala:76) > at > org.apache.flink.table.planner.plan.rules.logical.JoinConditionTypeCoerceRule$$anonfun$onMatch$1.apply(JoinConditionTypeCoerceRule.scala:65) > at scala.collection.Iterator$class.foreach(Iterator.scala:891) > at scala.collection.AbstractIterator.foreach(Iterator.scala:1334) > at scala.collection.IterableLike$class.foreach(IterableLike.scala:72) > at scala.collection.AbstractIterable.foreach(Iterable.scala:54) > at > org.apache.flink.table.planner.plan.rules.logical.JoinConditionTypeCoerceRule.onMatch(JoinConditionTypeCoerceRule.scala:65) > at > org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:328) > at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:562) > at > org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:427) > at > org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:264) > at > org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127) > at > org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:223) > at > org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:210) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram$$anonfun$optimize$1$$anonfun$apply$1.apply(FlinkGroupProgram.scala:63) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram$$anonfun$optimize$1$$anonfun$apply$1.apply(FlinkGroupProgram.scala:60) > at > scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157) > at > scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157) > at scala.collection.Iterator$class.foreach(Iterator.scala:891) > at scala.collection.AbstractIterator.foreach(Iterator.scala:1334) > at scala.collection.IterableLike$class.foreach(IterableLike.scala:72) > at scala.collection.AbstractIterable.foreach(Iterable.scala:54) > at > scala.collection.TraversableOnce$class.foldLeft(TraversableOnce.scala:157) > at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram$$anonfun$optimize$1.apply(FlinkGroupProgram.scala:60) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram$$anonfun$optimize$1.apply(FlinkGroupProgram.scala:55) > at > scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157) > at > scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157) > at scala.collection.immutable.Range.foreach(Range.scala:160) > at > scala.collection.TraversableOnce$class.foldLeft(TraversableOnce.scala:157) > at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram$$anonfun$optimize$1.apply(FlinkChainedProgram.scala:62) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram$$anonfun$optimize$1.apply(FlinkChainedProgram.scala:58) > at > scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157) > at > scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157) > at scala.collection.Iterator$class.foreach(Iterator.scala:891) > at scala.collection.AbstractIterator.foreach(Iterator.scala:1334) > at scala.collection.IterableLike$class.foreach(IterableLike.scala:72) > at scala.collection.AbstractIterable.foreach(Iterable.scala:54) > at > scala.collection.TraversableOnce$class.foldLeft(TraversableOnce.scala:157) > at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104) > at > org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:57) > at > org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:164) > at > org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:80) > at > org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77) > at > org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:279) > at > org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:164) > at > org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.toDataStream(StreamTableEnvironmentImpl.java:331) > at > org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl.toRetractStream(StreamTableEnvironmentImpl.java:307) > at > com.akulaku.blade.source.BladeKafkaJoinCodeTest.eventDeviceInfo(BladeKafkaJoinCodeTest.java:265) > at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) > at > sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) > at > sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) > at java.lang.reflect.Method.invoke(Method.java:498) > at > org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50) > at > org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12) > at > org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47) > at > org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17) > at > org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) > at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325) > at > org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78) > at > org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57) > at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290) > at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71) > at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288) > at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58) > at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268) > at org.junit.runners.ParentRunner.run(ParentRunner.java:363) > at org.junit.runner.JUnitCore.run(JUnitCore.java:137) > at > com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68) > at > com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:33) > at > com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:230) > at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:58) > > >Process finished with exit code 255 > > >hailongwang <[email protected]> 于2020年10月19日周一 下午8:35写道： > >> Hi Dream， >> 可以分享下你完整的程序吗，我感觉这个是因为 JOIN ON 条件上类型不一致引起的，可以分享下你完整的程序看下。 >> Best， >> Hailong Wang >> >> 在 2020-10-19 09:50:33，"Dream-底限" <[email protected]> 写道： >> >hi、我这面正在将flinktable转换为datastream，现在抛出如下异常，貌似是个bug。。。 >> > >> >table.printSchema(); >> >streamTableEnv.toRetractStream(table, >> >> >Types.ROW(TypeConversions.fromDataTypeToLegacyInfo(table.getSchema().getFieldDataTypes()))).print(); >> > >> > >> > >> >root >> > |-- register_id: BIGINT >> > |-- asi_uid: BIGINT >> > |-- person_uuid: BIGINT >> > |-- app_id: BIGINT >> > |-- country_id: BIGINT >> > |-- channel_id: STRING >> > |-- device_id: STRING >> > |-- adjust_id: STRING >> > |-- google_adid: STRING >> > |-- referrer: BIGINT >> > |-- login_pwd: STRING >> > |-- sync_data_flag: INT >> > |-- register_phone_number: STRING >> > |-- device_type: INT >> > |-- imei: STRING >> > |-- device_model: STRING >> > |-- os_version: STRING >> > |-- app_name: STRING >> > |-- app_version: STRING >> > |-- app_package_name: STRING >> > |-- network_type: STRING >> > |-- wifi_mac: STRING >> > |-- longitude: DECIMAL(38, 18) >> > |-- latitude: DECIMAL(38, 18) >> > |-- geo_hash7: STRING >> > |-- ip: STRING >> > |-- register_time: BIGINT >> > |-- etl_time: BIGINT NOT NULL >> > >> > >> >org.apache.flink.table.api.TableException: BIGINT and >> >VARCHAR(2147483647) does not have common type now >>
