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 "1129656513@qq.com" <11...@qq.com> on 2020/07/22 02:08:56 UTC
flinksql1.11中主键声明的问题
hi：
我在使用pyflink1.11过程中，使用flinksql维表时声明了主键primary key 但是还是会报错说我没有用声明主键，另外，当我使用inner join代替left join就不会有这个问题，请问这是什么问题
下面我附录了报错信息和代码。谢谢！
报错附录
Traceback (most recent call last):
File "/usr/local/lib/python3.7/site-packages/pyflink/util/exceptions.py", line 147, in deco
return f(*a, **kw)
File "/usr/local/lib/python3.7/site-packages/py4j/protocol.py", line 328, in get_return_value
format(target_id, ".", name), value)
py4j.protocol.Py4JJavaError: An error occurred while calling o5.execute.
: org.apache.flink.table.api.TableException: UpsertStreamTableSink requires that Table has a full primary keys if it is updated.
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:93)
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:48)
at org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlan(StreamExecLegacySink.scala:48)
at org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:67)
at org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:66)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
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.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at org.apache.flink.table.planner.delegation.StreamPlanner.translateToPlan(StreamPlanner.scala:66)
at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:166)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1248)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.translateAndClearBuffer(TableEnvironmentImpl.java:1240)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.execute(TableEnvironmentImpl.java:1197)
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.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:748)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "mysql_join.py", line 90, in <module>
from_kafka_to_kafka_demo()
File "mysql_join.py", line 22, in from_kafka_to_kafka_demo
st_env.execute("2-from_kafka_to_kafka")
File "/usr/local/lib/python3.7/site-packages/pyflink/table/table_environment.py", line 1057, in execute
return JobExecutionResult(self._j_tenv.execute(job_name))
File "/usr/local/lib/python3.7/site-packages/py4j/java_gateway.py", line 1286, in __call__
answer, self.gateway_client, self.target_id, self.name)
File "/usr/local/lib/python3.7/site-packages/pyflink/util/exceptions.py", line 154, in deco
raise exception_mapping[exception](s.split(': ', 1)[1], stack_trace)
pyflink.util.exceptions.TableException: 'UpsertStreamTableSink requires that Table has a full primary keys if it is updated.'
代码附录
from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
from pyflink.table import StreamTableEnvironment, DataTypes, EnvironmentSettings,DataTypes, CsvTableSource, CsvTableSink
from pyflink.table.descriptors import Schema, Kafka, Json, Rowtime
from pyflink.table.window import Tumble
def from_kafka_to_kafka_demo():
# use blink table planner
env = StreamExecutionEnvironment.get_execution_environment()
env.set_stream_time_characteristic(TimeCharacteristic.EventTime)
env_settings = EnvironmentSettings.Builder().use_blink_planner().build()
st_env = StreamTableEnvironment.create(stream_execution_environment=env,environment_settings=env_settings)
# register source and sink
register_rides_source(st_env)
register_rides_sink(st_env)
register_mysql_source(st_env)
st_env.sql_update("insert into flink_result select cast(t1.id as int) as id,cast(t2.type as varchar),cast( t1.time1 as bigint) as rowtime from source1 t1 left join dim_mysql t2 on t1.type=cast(t2.id as varchar) ")
st_env.execute("2-from_kafka_to_kafka")
def register_rides_source(st_env):
source_ddl = \
"""
create table source1(
id int,
time1 varchar ,
type string
) with (
'connector.type' = 'kafka',
'connector.topic' = 'tp1',
'connector.startup-mode' = 'latest-offset',
'connector.properties.bootstrap.servers' = 'localhost:9092',
'connector.properties.zookeeper.connect' = 'localhost:2181',
'format.type' = 'json',
'connector.version' = 'universal',
'update-mode' = 'append'
)
"""
st_env.sql_update(source_ddl)
def register_mysql_source(st_env):
source_ddl = \
"""
CREATE TABLE dim_mysql (
id int, --
type varchar --
) WITH (
'connector.type' = 'jdbc',
'connector.url' = 'jdbc:mysql://localhost:3390/test',
'connector.table' = 'flink_test',
'connector.driver' = 'com.mysql.cj.jdbc.Driver',
'connector.username' = '***',
'connector.password' = '***',
'connector.lookup.cache.max-rows' = '5000',
'connector.lookup.cache.ttl' = '1min',
'connector.lookup.max-retries' = '3'
)
"""
st_env.sql_update(source_ddl)
def register_rides_sink(st_env):
sink_ddl = \
"""
CREATE TABLE flink_result (
id int,
type varchar,
rtime bigint,
primary key(id) NOT ENFORCED
) WITH (
'connector.type' = 'jdbc',
'connector.url' = 'jdbc:mysql://localhost:3390/test',
'connector.table' = 'flink_result',
'connector.driver' = 'com.mysql.cj.jdbc.Driver',
'connector.username' = '***',
'connector.password' = '***',
'connector.write.flush.max-rows' = '5000',
'connector.write.flush.interval' = '2s',
'connector.write.max-retries' = '3'
)
"""
st_env.sql_update(sink_ddl)
if __name__ == '__main__':
from_kafka_to_kafka_demo()
初学者
琴师
回复: Re: flinksql1.11中主键声明的问题
Posted by 琴师 <11...@qq.com>.
你好：
下面是我的代码，我用的版本是1.11.0，数据库是TIDB，我跑的是demo数据，维表只有两行。
我的输入流如下，每秒新增一条写入到kafka
topic = 'tp1'
for i in range(1,10000) :
stime=datetime.datetime.now().strftime('%Y%m%d%H%M%S')
msg = {}
msg['id']= i
msg['time1']= stime
msg['type']=1
print(msg)
send_msg(topic, msg)
time.sleep(1)
{'id': 1, 'time1': '20200722140624', 'type': 1}
{'id': 2, 'time1': '20200722140625', 'type': 1}
{'id': 3, 'time1': '20200722140626', 'type': 1}
{'id': 4, 'time1': '20200722140627', 'type': 1}
{'id': 5, 'time1': '20200722140628', 'type': 1}
{'id': 6, 'time1': '20200722140629', 'type': 1}
{'id': 7, 'time1': '20200722140631', 'type': 1}
{'id': 8, 'time1': '20200722140632', 'type': 1}
维表数据如下
id type
2 err
1 err
我在程序正常期间更新了维表，但是后续输出的结果显示维表还是之前的缓存数据，事实上已经远远大于超时时间了，甚至我停下输入流，直到达到超时时间后再次输入，新的结果还是输出旧的维表数据
from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
from pyflink.table import StreamTableEnvironment, DataTypes, EnvironmentSettings,DataTypes, CsvTableSource, CsvTableSink
from pyflink.table.descriptors import Schema, Kafka, Json, Rowtime
from pyflink.table.window import Tumble
def from_kafka_to_kafka_demo():
# use blink table planner
env = StreamExecutionEnvironment.get_execution_environment()
env.set_stream_time_characteristic(TimeCharacteristic.EventTime)
env_settings = EnvironmentSettings.Builder().use_blink_planner().build()
st_env = StreamTableEnvironment.create(stream_execution_environment=env,environment_settings=env_settings)
# register source and sink
register_rides_source(st_env)
register_rides_sink(st_env)
register_mysql_source(st_env)
st_env.sql_update("insert into flink_result select cast(t1.id as int) as id,cast(t2.type as varchar),cast( t1.time1 as bigint) as rowtime from source1 t1 left join dim_mysql t2 on t1.type=cast(t2.id as varchar) ")
st_env.execute("2-from_kafka_to_kafka")
def register_rides_source(st_env):
source_ddl = \
"""
create table source1(
id int,
time1 varchar ,
type string
) with (
'connector' = 'kafka',
'topic' = 'tp1',
'scan.startup.mode' = 'latest-offset',
'properties.bootstrap.servers' = 'localhost:9092',
'format' = 'json'
)
"""
st_env.sql_update(source_ddl)
def register_mysql_source(st_env):
source_ddl = \
"""
CREATE TABLE dim_mysql (
id int, --
type varchar --
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://localhost:3390/test',
'table-name' = 'flink_test',
'driver' = 'com.mysql.cj.jdbc.Driver',
'username' = '***',
'password' = '***',
'lookup.cache.max-rows' = '5000',
'lookup.cache.ttl' = '1s',
'lookup.max-retries' = '3'
)
"""
st_env.sql_update(source_ddl)
def register_rides_sink(st_env):
sink_ddl = \
"""
CREATE TABLE flink_result (
id int,
type varchar,
rtime bigint,
primary key(id) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://localhost:3390/test',
'table-name' = 'flink_result',
'driver' = 'com.mysql.cj.jdbc.Driver',
'username' = '***',
'password' = '***',
'sink.buffer-flush.max-rows' = '5000',
'sink.buffer-flush.interval' = '2s',
'sink.max-retries' = '3'
)
"""
st_env.sql_update(sink_ddl)
if __name__ == '__main__':
from_kafka_to_kafka_demo()
初学者
PyFlink爱好者
琴师
发件人： Leonard Xu
发送时间： 2020-07-22 15:05
收件人： user-zh
主题： Re: flinksql1.11中主键声明的问题
Hi,
我试了下应该是会更新缓存的，你有能复现的例子吗？
祝好
> 在 2020年7月22日，14:50，奇怪的不朽琴师 <11...@qq.com> 写道：
>
> 你好：
>
>
> 可能是我描述的不清楚，我了解这个机制，我的意思维表更新后，即便已经达到了超时的时间，新的输出结果还是用维表历史缓存数据，
> 我感觉上是维表没有刷新缓存，但是我不知道这为什么。
>
>
> 谢谢
>
>
> ------------------ 原始邮件 ------------------
> 发件人: "user-zh" <xbjtdcq@gmail.com>;
> 发送时间: 2020年7月22日(星期三) 下午2:42
> 收件人: "user-zh"<user-zh@flink.apache.org>;
>
> 主题: Re: flinksql1.11中主键声明的问题
>
>
>
> Hello
> 你说的输出结果更新，是指之前关联的维表时老数据，过了一段时间，这个数据变，之前输出的历史也希望更新吗？维表join的实现，只有事实表中才会有retract消息才会更新，才会传递到下游，维表的数据是事实表
> 去look up该表时的数据，维表的更新是不会retract之前的历史记录的。
>
> 祝好
> Leonard Xu
>
>
> > 在 2020年7月22日，14:13，1129656513@qq.com 写道：
> >
> > 输出结果仍然没有被更新
Re: flinksql1.11中主键声明的问题
Posted by Leonard Xu <xb...@gmail.com>.
Hi,
我试了下应该是会更新缓存的，你有能复现的例子吗？
祝好
> 在 2020年7月22日，14:50，奇怪的不朽琴师 <11...@qq.com> 写道：
>
> 你好：
>
>
> 可能是我描述的不清楚，我了解这个机制，我的意思维表更新后，即便已经达到了超时的时间，新的输出结果还是用维表历史缓存数据，
> 我感觉上是维表没有刷新缓存，但是我不知道这为什么。
>
>
> 谢谢
>
>
> ------------------ 原始邮件 ------------------
> 发件人: "user-zh" <xbjtdcq@gmail.com>;
> 发送时间: 2020年7月22日(星期三) 下午2:42
> 收件人: "user-zh"<user-zh@flink.apache.org>;
>
> 主题: Re: flinksql1.11中主键声明的问题
>
>
>
> Hello
> 你说的输出结果更新，是指之前关联的维表时老数据，过了一段时间，这个数据变，之前输出的历史也希望更新吗？维表join的实现，只有事实表中才会有retract消息才会更新，才会传递到下游，维表的数据是事实表
> 去look up该表时的数据，维表的更新是不会retract之前的历史记录的。
>
> 祝好
> Leonard Xu
>
>
> > 在 2020年7月22日，14:13，1129656513@qq.com 写道：
> >
> > 输出结果仍然没有被更新
回复： flinksql1.11中主键声明的问题
Posted by 奇怪的不朽琴师 <11...@qq.com>.
你好：
可能是我描述的不清楚，我了解这个机制，我的意思维表更新后，即便已经达到了超时的时间，新的输出结果还是用维表历史缓存数据，
我感觉上是维表没有刷新缓存，但是我不知道这为什么。
谢谢
------------------ 原始邮件 ------------------
发件人: "user-zh" <xbjtdcq@gmail.com>;
发送时间: 2020年7月22日(星期三) 下午2:42
收件人: "user-zh"<user-zh@flink.apache.org>;
主题: Re: flinksql1.11中主键声明的问题
Hello
你说的输出结果更新，是指之前关联的维表时老数据，过了一段时间，这个数据变，之前输出的历史也希望更新吗？维表join的实现，只有事实表中才会有retract消息才会更新，才会传递到下游，维表的数据是事实表
去look up该表时的数据，维表的更新是不会retract之前的历史记录的。
祝好
Leonard Xu
> 在 2020年7月22日，14:13，1129656513@qq.com 写道：
>
> 输出结果仍然没有被更新
Re: flinksql1.11中主键声明的问题
Posted by Leonard Xu <xb...@gmail.com>.
Hello
你说的输出结果更新，是指之前关联的维表时老数据，过了一段时间，这个数据变，之前输出的历史也希望更新吗？维表join的实现，只有事实表中才会有retract消息才会更新，才会传递到下游，维表的数据是事实表
去look up该表时的数据，维表的更新是不会retract之前的历史记录的。
祝好
Leonard Xu
> 在 2020年7月22日，14:13，1129656513@qq.com 写道：
>
> 输出结果仍然没有被更新
回复: Re: flinksql1.11中主键声明的问题
Posted by "1129656513@qq.com" <11...@qq.com>.
您好：
非常感谢您的建议，我已经成功解决了这个问题，但是我又发现了一个新的问题，我这里设置的超时时间是一分钟或者超时行数是5000行，
我在这期间更新了维表数据，但是我发现已经超过了超时时间，输出结果仍然没有被更新，是我理解的有问题么？
我尝试了停止输入流数据直到达到超时时间后仍然没有更新维表，除非停止整个程序，否则我的维表数据都不会被更新。
请问这个问题有解决的办法么？
def register_mysql_source(st_env):
source_ddl = \
"""
CREATE TABLE dim_mysql (
id int, --
type varchar --
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://localhost:3390/test',
'table-name' = 'flink_test',
'driver' = 'com.mysql.cj.jdbc.Driver',
'username' = '****',
'password' = '****',
'lookup.cache.max-rows' = '5000',
'lookup.cache.ttl' = '1s',
'lookup.max-retries' = '3'
)
"""
st_env.sql_update(source_ddl)
感谢！
琴师
发件人： Leonard Xu
发送时间： 2020-07-22 10:54
收件人： user-zh
主题： Re: flinksql1.11中主键声明的问题
Hi,
你这还是connector的with参数里不是新 connector的写法[1]，会走到老代码，老代码不支持声明PK的。
在老代码里，PK是通过query推导的，你用inner join替换left join后，应该是能够推断出PK了，所以没有报错。
我理解你把connector的with参数更新成新的就解决问题了。
Best
Leonard Xu
[1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html#connector-options <https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html#connector-options>
>
> def register_rides_source(st_env):
> source_ddl = \
> """
> create table source1(
> id int,
> time1 varchar ,
> type string
> ) with (
> 'connector.type' = 'kafka',
> 'connector.topic' = 'tp1',
> 'connector.startup-mode' = 'latest-offset',
> 'connector.properties.bootstrap.servers' = 'localhost:9092',
> 'connector.properties.zookeeper.connect' = 'localhost:2181',
> 'format.type' = 'json',
> 'connector.version' = 'universal',
> 'update-mode' = 'append'
> )
> “""
Re: flinksql1.11中主键声明的问题
Posted by Leonard Xu <xb...@gmail.com>.
Hi,
你这还是connector的with参数里不是新 connector的写法[1]，会走到老代码，老代码不支持声明PK的。
在老代码里，PK是通过query推导的，你用inner join替换left join后，应该是能够推断出PK了，所以没有报错。
我理解你把connector的with参数更新成新的就解决问题了。
Best
Leonard Xu
[1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html#connector-options <https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html#connector-options>
>
> def register_rides_source(st_env):
> source_ddl = \
> """
> create table source1(
> id int,
> time1 varchar ,
> type string
> ) with (
> 'connector.type' = 'kafka',
> 'connector.topic' = 'tp1',
> 'connector.startup-mode' = 'latest-offset',
> 'connector.properties.bootstrap.servers' = 'localhost:9092',
> 'connector.properties.zookeeper.connect' = 'localhost:2181',
> 'format.type' = 'json',
> 'connector.version' = 'universal',
> 'update-mode' = 'append'
> )
> “""