You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by Dream-底限 <zh...@akulaku.com> on 2020/10/19 02:50:33 UTC
flink table转datastream失败
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
Re: Re: flink table转datastream失败
Posted by Dream-底限 <zh...@akulaku.com>.
hi、
是的，类型部分不匹配，类型改完之后程序运行正常了，感谢
hailongwang <18...@163.com> 于2020年10月20日周二 下午4:13写道：
> Hi,
> 我看其中一个 condition 是 `t1.uid = t2.refer_id`
> 其中 uid 是 bigint 类型，refer_id 是 varchar 类型。
> 你再确认下？
>
>
> Best,
> Hailong Wang
>
>
>
>
> At 2020-10-20 08:55:34, "Dream-底限" <zh...@akulaku.com> 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 <18...@163.com> 于2020年10月19日周一 下午8:35写道：
> >
> >> Hi Dream，
> >> 可以分享下你完整的程序吗，我感觉这个是因为 JOIN ON 条件上类型不一致引起的，可以分享下你完整的程序看下。
> >> Best，
> >> Hailong Wang
> >>
> >> 在 2020-10-19 09:50:33，"Dream-底限" <zh...@akulaku.com> 写道：
> >> >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
> >>
>
Re:Re: flink table转datastream失败
Posted by hailongwang <18...@163.com>.
Hi,
我看其中一个 condition 是 `t1.uid = t2.refer_id`
其中 uid 是 bigint 类型，refer_id 是 varchar 类型。
你再确认下？
Best,
Hailong Wang
At 2020-10-20 08:55:34, "Dream-底限" <zh...@akulaku.com> 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 <18...@163.com> 于2020年10月19日周一 下午8:35写道：
>
>> Hi Dream，
>> 可以分享下你完整的程序吗，我感觉这个是因为 JOIN ON 条件上类型不一致引起的，可以分享下你完整的程序看下。
>> Best，
>> Hailong Wang
>>
>> 在 2020-10-19 09:50:33，"Dream-底限" <zh...@akulaku.com> 写道：
>> >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
>>
Re: flink table转datastream失败
Posted by Dream-底限 <zh...@akulaku.com>.
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 <18...@163.com> 于2020年10月19日周一 下午8:35写道：
> Hi Dream，
> 可以分享下你完整的程序吗，我感觉这个是因为 JOIN ON 条件上类型不一致引起的，可以分享下你完整的程序看下。
> Best，
> Hailong Wang
>
> 在 2020-10-19 09:50:33，"Dream-底限" <zh...@akulaku.com> 写道：
> >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
>
Re:flink table转datastream失败
Posted by hailongwang <18...@163.com>.
Hi Dream，
可以分享下你完整的程序吗，我感觉这个是因为 JOIN ON 条件上类型不一致引起的，可以分享下你完整的程序看下。
Best，
Hailong Wang
在 2020-10-19 09:50:33，"Dream-底限" <zh...@akulaku.com> 写道：
>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