> 在 2020年5月20日，15:27，Leonard Xu <
[hidden email]> 写道：
>
> Hi，guaishushu
>
> 先说声抱歉邮件回复晚了，过了下你的sql，问题是1.10 中 对于upsertSink的primary key是通过query来推断的，部分query是推断不出来的，你的query刚好推断
> 不出来PK的，所以会提示：Exception in thread "main" org.apache.flink.table.api.TableException: UpsertStreamTableSink requires that Table has a full primary keys if it is updated.
>
> 在1.10的临时解决方法是加一层group by，使得query可以推断出 primary key：
> INSERT INTO test_mysql_2
> SELECT order_id,rss, start_time FROM(
>> SELECT vid,rss, start_time FROM (
>> SELECT vid,rss, start_time,
>> ROW_NUMBER() OVER (PARTITION BY start_time ORDER BY rss desc) AS rownum
>> FROM (
>> SELECT vid,
>> DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '5' MINUTE),'yyyy-MM-dd HH:00') AS start_time,
>> SUM(response_size) AS rss
>> FROM user_log
>> GROUP BY vid, TUMBLE(rowtime, INTERVAL '5' MINUTE)
>> )
>> )
>> WHERE rownum <= 10
>
>
> ) group by order_id,rss, start_time
>
> 这确实是不完善的地方，即将到来的1.11中已经有了优雅的解决方案，可以在sink的表中声明primary key 而不是通过query推断PK[1].
>
>
> Best,
> Leonard
> [1]
https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSinkITCase.java#L184 <
https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSinkITCase.java#L184>
>
>
>> 在 2020年5月11日，10:26，
[hidden email] <mailto:
[hidden email]> 写道：
>>
>> CREATE TABLE test_mysql_2 (
>> vid string,
>> rss BIGINT,
>> start_time string
>> ) with (
>> 'connector.type' = 'jdbc',
>> 'connector.url' = 'jdbc:mysql://xxx/xxx?useUnicode=true&characterEncoding=utf8&autoReconnect=true&failOverReadOnly=false' <mysql://xxx/xxx?useUnicode=true&characterEncoding=utf8&autoReconnect=true&failOverReadOnly=false'>,
>> 'connector.username' = 'xxx',
>> 'connector.password' = 'xxx',
>> 'connector.table' = 'task_flink_table_3',
>> 'connector.write.flush.max-rows' = '100'
>> );
>>
>>
>>
>>
>> INSERT INTO test_mysql_2
>> SELECT vid,rss, start_time FROM (
>> SELECT vid,rss, start_time,
>> ROW_NUMBER() OVER (PARTITION BY start_time ORDER BY rss desc) AS rownum
>> FROM (
>> SELECT vid,
>> DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '5' MINUTE),'yyyy-MM-dd HH:00') AS start_time,
>> SUM(response_size) AS rss
>> FROM user_log
>> GROUP BY vid, TUMBLE(rowtime, INTERVAL '5' MINUTE)
>> )
>> )
>> WHERE rownum <= 10;
>>
>>
[hidden email] <mailto:
[hidden email]>
>>
>> 发件人： Leonard Xu <mailto:
[hidden email]>
>> 发送时间： 2020-05-09 14:15
>> 收件人： user-zh <mailto:
[hidden email]>
>> 主题： Re: Flink-1.10-SQL TopN语法问题
>> Hi
>>
>> 图挂了，可以用图床工具链接出来，或者可以把query贴出来看看嘛？
>>
>> Best,
>> Leonard Xu
>>
>> > 在 2020年5月9日，13:51，
[hidden email] <mailto:
[hidden email]> 写道：
>> >
>> > hi 大家，我在使用TopN语法，往mysql写数据时，发现必须将rownum设置为数据库的主键，要不然会报错，有人遇到吗
>> >
>> >
[hidden email] <mailto:
[hidden email]> <mailto:
[hidden email] <mailto:
[hidden email]>>
>
Locked
