-建表语法如下
String kafka = "CREATE TABLE `电话` " +
"(`rowID` VARCHAR(255),`名称` STRING,`手机` VARCHAR(255),`座机` VARCHAR(255), " +
" PRIMARY KEY (`rowID`) NOT ENFORCED ) " +
" WITH " +
"('connector' = 'jdbc', " +
" 'driver' = 'com.mysql.cj.jdbc.Driver', " +
" 'url' =
>> Hi
> >> >>
> >> >> 如果join计算的关联字段里没有主键，在将join结果直接写入sink表时，像上面RS提到的，需要自己再增加一次shuffle操作
> >> >>
> >> >> Best,
> >> >> Shammon
> >> >>
> >> >>
> >> >> On Sun, Feb 19, 2023 at 1:43 PM
>> > Hi，
>> >> > connector里面配置的主键控制是写入存储的，有些存储在写入数据的时候，能根据主键自动更新去重
>> >> > 所以我感觉，你这里的应该是想在计算的时候shuffle（写入之前），你应该需要先执行一个 group by 主键，然后再执行insert
>> into
>> >> >
>> >> >
>> >> > Thanks
>> >> >
>
先执行一个 group by 主键，然后再执行insert
> into
> >> >
> >> >
> >> > Thanks
> >> >
> >> >
> >> >
> >> > 在 2023-02-17 15:56:51，"casel.chen" 写道：
> >> > >作业场景是kafka cdc数据源关联几张redis维表再和其他流表进行双流regular i
nnector里面配置的主键控制是写入存储的，有些存储在写入数据的时候，能根据主键自动更新去重
> >> > 所以我感觉，你这里的应该是想在计算的时候shuffle（写入之前），你应该需要先执行一个 group by 主键，然后再执行insert
> into
> >> >
> >> >
> >> > Thanks
> >> >
> >> >
> >> >
> >> > 在 2023-02-17 1
>
>> >
>> > 在 2023-02-17 15:56:51，"casel.chen" 写道：
>> > >作业场景是kafka cdc数据源关联几张redis维表再和其他流表进行双流regular inner
>> > join，最后将打宽表写入mongodb。使用的是flink 1.13.2 sql模式。开了debug日志。
>> > >测试下来发现相同主键的记录在不同的taskmanager上打了日志（我是在Sink
>> > Function的
One question as title: Whether Flink SQL window operations support
> "Allow Lateness and SideOutput"?
>
> Just as supported in Datastream api (allowedLateness
> and sideOutputLateData) like:
>
> SingleOutputStreamOperator<
Hi dear engineers,
One question as title: Whether Flink SQL window operations support "Allow
Lateness and SideOutput"?
Just as supported in Datastream api (allowedLateness and sideOutputLateData)
like:
SingleOutputStreamOperator<>sumStream = dataStream.ke
Flink SQL作业示意如下：
create table user_source_table (
id BIGINT NOT NULL PRIMARY KEY NOT ENFORCED,
name STRING,
dept_id BIGINT NOT NULL,
proctime AS PROCTIME()
) with (
'connector' = 'kafka',
'format' = 'canal-json',
...
);
create table department_dim_table (
id BIGINT NOT NULL
nk 1.13.2 sql模式。开了debug日志。
>>测试下来发现相同主键的记录在不同的taskmanager上打了日志（我是在Sink
>>Function的invoke方法打的日志），该行为导致最终结果表数据不正确。
>>
>>
>>请问：
>>flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？
>>是因为flink版本旧不支持吗？从flink哪个版本开始支持的呢？
>>我理解flink
>>sql结果表上定义主键的目的就是期望按主键进行shuffle，确保相同主键的数据被同一个taskmanager处理以确保正确的变更顺序，不知道我理解得对不对。
>>
t; >
> >
> > Thanks
> >
> >
> >
> > 在 2023-02-17 15:56:51，"casel.chen" 写道：
> > >作业场景是kafka cdc数据源关联几张redis维表再和其他流表进行双流regular inner
> > join，最后将打宽表写入mongodb。使用的是flink 1.13.2 sql模式。开了debug日志。
> > >测试下来发现相同主键的记录在不同的task
t;
>
>
> 在 2023-02-17 15:56:51，"casel.chen" 写道：
> >作业场景是kafka cdc数据源关联几张redis维表再和其他流表进行双流regular inner
> join，最后将打宽表写入mongodb。使用的是flink 1.13.2 sql模式。开了debug日志。
> >测试下来发现相同主键的记录在不同的taskmanager上打了日志（我是在Sink
> Function的invoke方法打的日志），该行为导致最终结果表数据不正确。
> >
>
ebug日志。
>测试下来发现相同主键的记录在不同的taskmanager上打了日志（我是在Sink
>Function的invoke方法打的日志），该行为导致最终结果表数据不正确。
>
>
>请问：
>flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？
>是因为flink版本旧不支持吗？从flink哪个版本开始支持的呢？
>我理解flink
>sql结果表上定义主键的目的就是期望按主键进行shuffle，确保相同主键的数据被同一个taskmanager处理以确保正确的变更顺序，不知道我理解得对不对。
>
作业场景是kafka cdc数据源关联几张redis维表再和其他流表进行双流regular inner
join，最后将打宽表写入mongodb。使用的是flink 1.13.2 sql模式。开了debug日志。
测试下来发现相同主键的记录在不同的taskmanager上打了日志（我是在Sink
Function的invoke方法打的日志），该行为导致最终结果表数据不正确。
请问：
flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？
是因为flink版本旧不支持吗？从flink哪个版本开始支持的呢？
我理解flink
sql结果表上定
Hi
可以考虑将这个function打入到udf包里，在自定义的udf里直接调用？
On Wed, Feb 15, 2023 at 4:29 PM 723849736 <723849...@qq.com.invalid> wrote:
> 大家好，
>
> 我在用flink sql的时候有一个场景，就是需要对数组中的某一列做变换，类似于spark sql中的tranform函数
>
>
> https://spark.apache.org/docs/latest/api/sql/index.html#transform
>
&
你好，可以贴一下详细的异常栈吗？
可以在 FLINK_HOME/log/ 下面看到
Best regards,
Yuxia
- 原始邮件 -
发件人: "aiden" <18765295...@163.com>
收件人: "user-zh"
发送时间: 星期一, 2023年 2 月 06日 下午 4:44:02
主题: Flink SQL使用hive dialect异常
HI
我在使用Flink SQL Client开发hive时遇到一个问题，在设置set table.sql-dialect=hive
qq.com.invalid> 于2023年1月31日周二 17:22写道：
>
>> 你可以测试不写入clickhouse是否还存在反压，如果不是因为写入瓶颈的话就从你的处理逻辑优化了
>>
>>
>> 发件人: lxk
>> 发送时间: 2023年1月31日 15:16
>> 收件人: user-zh@flink.apache.org
>> 主题: Flink SQL 如何优化以及处理反压
>>
>> Flink版本：1.16.0
>> 目前在使用Flink
最好先找到导致下游处理过慢的瓶颈算子，适当扩大一下并发。如果还不行，看下jstack的情况，可能需要调整逻辑。
Best regards,
Weijie
ssmq <374060...@qq.com.invalid> 于2023年1月31日周二 17:22写道：
> 你可以测试不写入clickhouse是否还存在反压，如果不是因为写入瓶颈的话就从你的处理逻辑优化了
>
>
> 发件人: lxk
> 发送时间: 2023年1月31日 15:16
> 收件人: user-zh@flink.apache.org
> 主题: Flin
你可以测试不写入clickhouse是否还存在反压，如果不是因为写入瓶颈的话就从你的处理逻辑优化了
发件人: lxk
发送时间: 2023年1月31日 15:16
收件人: user-zh@flink.apache.org
主题: Flink SQL 如何优化以及处理反压
Flink版本：1.16.0
目前在使用Flink SQL进行多流关联，并写入Clickhouse中
具体代码如下：
select \
header.id as id, \
LAST_VALUE(header.order_status), \
LAST_VALUE(header.customer_id
Flink版本：1.16.0
目前在使用Flink SQL进行多流关联，并写入Clickhouse中
具体代码如下：
select \
header.id as id, \
LAST_VALUE(header.order_status), \
LAST_VALUE(header.customer_id), \
LAST_VALUE(header.shop_id), \
LAST_VALUE(header.parent_order_id), \
LAST_VALUE(header.order_at), \
LAST_VALUE(header.pay_at), \
LAST_VALUE
(HadoopFileSystem.java:165)
[flink-dist-1.16.0.jar:1.16.0]
at org.apache.flink.table.client.SqlClient.start(SqlClient.java:95)
[flink-sql-client-1.16.0.jar:1.16.0]
at
org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:187)
[flink-sql-client-1.16.0.jar:1.16.0
Hi, casel
Flink 本身会采集 JVM 层面的资源使用量，详情可以参考官方文档
https://nightlies.apache.org/flink/flink-docs-master/docs/ops/metrics/#cpu
除此之外，可以看下 Kubernetes 是否部署了相关的资源监控服务，可以从 Pod 层面查看资源使用率。
Best,
Weihua
On Tue, Jan 17, 2023 at 11:41 AM casel.chen wrote:
> 我们flink
> sql作业跑在k8s上，但发现k8s集群整体资源使用率并不高，例如
我们flink sql作业跑在k8s上，但发现k8s集群整体资源使用率并不高，例如请求内存占总内存89.28%，但实际使用内存占总内存只有66.38%。
现在想排查出哪些作业过度申请资源，有什么办法或直接的metrics可以监控flink sql作业实现k8s资源使用率么？谢谢！
看过了，不支持http source table，而且即使http lookup table也不支持map数据类型
在 2022-12-19 14:51:42，"Weihua Hu" 写道：
>Hi, 你可以尝试使用独立开源的 http connector
>
>https://github.com/getindata/flink-http-connector
>
>Best,
>Weihua
>
>
>On Sat, Dec 17, 2022 at 10:21 AM ca
retrieving result.
at
org.apache.flink.table.client.gateway.local.result.CollectResultBase$ResultRetrievalThread.run(CollectResultBase.java:79)
~[flink-sql-client-1.16.0.jar:1.16.0]
Caused by: java.lang.RuntimeException: Failed to fetch n
Hi, 你可以尝试使用独立开源的 http connector
https://github.com/getindata/flink-http-connector
Best,
Weihua
On Sat, Dec 17, 2022 at 10:21 AM casel.chen wrote:
> 我想开发一个Flink SQL Http Connector，遇到用户添加自定义请求头Headers问题，如何在connector
> options中支持Map数据类型呢？
我想开发一个Flink SQL Http Connector，遇到用户添加自定义请求头Headers问题，如何在connector
options中支持Map数据类型呢？
chen 于2022年12月7日周三 11:52写道：
>
>> 有人能够解答一下吗？
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>> 在 2022-11-26 11:20:34，"casel.chen" 写道：
>> >双流关联场景下，流A数据到达后并不马上和流B进行关联，而是经过设置一段时间后再关联，这种场景下如何用flink
>> sql实现？如果当前不支持，需要怎样扩展flink sql呢？
>>
谢谢你给的建议，不过我们还没有升级到flink 1.16，目前在使用的是flink 1.15。
如果要使用flink sql来实现的话，是不是可以利用窗口去重来达到数据延迟关联的效果？
在每条数据到达后开一个10分钟累加窗口(step和size均为10分钟）根据key去重，在等待窗口结束之时输出的去重结果再跟维表进行lookup join
在 2022-12-07 13:33:50，"Lincoln Lee" 写道：
>双流 join 的场景下可以考虑关联条件引入时间属性，变成 interval join 可以实现一定
/dev/table/sql/queries/hints/#3-enable-delayed-retry-strategy-for-lookup
Best,
Lincoln Lee
casel.chen 于2022年12月7日周三 11:52写道：
> 有人能够解答一下吗？
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
> 在 2022-11-26 11:20:34，"casel.chen" 写道：
> >双流关联场
有人能够解答一下吗？
在 2022-11-26 11:20:34，"casel.chen" 写道：
>双流关联场景下，流A数据到达后并不马上和流B进行关联，而是经过设置一段时间后再关联，这种场景下如何用flink
>sql实现？如果当前不支持，需要怎样扩展flink sql呢？
维表流数据晚于主表流数据到达甚至可能到达不了，所以想设置个5分钟等待窗口，关联上正常处理，关联不上发到另一个kafka topic，这种场景使用flink
sql要如何实现？
hello, all
我通过flink sql去group by
uid计算滑动窗口，窗口大小一天，五分钟一个步长。譬如我uid有100w，那么五分钟就会往下游发送100w数据，如果是upsert-kafka，可能会发送200w数据（DELETE+INSERT）。
已经确定的是，大部分uid每五分钟计算出来的指标与前五分钟的一致，可能一天下来变化的uid只有几十w，有没有可能可以做一个判断，将要输出的指标与上一次的指标做一下判断，如果一致则不输出，如果不一致则输出？
可以考虑不用滑动窗口，但是每个uid变化后需要输出到下游，不变化则不需要输出。
Best wishes
el.chen |
>| 发送日期 | 2022年12月5日 18:24 |
>| 收件人 | user-zh@flink.apache.org |
>| 主题 | flink sql消费upsert-kafka源表如何指定从哪个位点开始消费？ |
>flink sql消费upsert-kafka源表如何指定从哪个位点开始消费？仿照kafka source表添加了 scan.startup.mode
>参数会报非法参数
hi
Upsert-kafka 不支持指定消费者位置，默认是从 earliest 位置开始消费的，你可以自己修改代码支持 scan.startup.mode 参数。
Best
JasonLee
回复的原邮件
| 发件人 | casel.chen |
| 发送日期 | 2022年12月5日 18:24 |
| 收件人 | user-zh@flink.apache.org |
| 主题 | flink sql消费upsert-kafka源表如何指定从哪个位点开始消费？ |
flink sql消费upsert-kafka源表如何指定从哪个位点开始消费？仿照
flink sql消费upsert-kafka源表如何指定从哪个位点开始消费？仿照kafka source表添加了 scan.startup.mode
参数会报非法参数
类方法？
>
>
>
>
>
>
>
>
>
>
>
> 在 2022-11-30 20:08:44，"Jiangang Liu" 写道：
> >Flink目前无法做到无缝升级，需要走stop-with-savepoint、start
> >job的流程，但是在这之间可以做一些优化来缩短恢复时间。比如，把新作业先启动起来，申请好资源，同时停掉老作业，将做好的savepoint用来触发新作业的执行。
> >
> >casel.chen 于2022年11月29日周
之间可以做一些优化来缩短恢复时间。比如，把新作业先启动起来，申请好资源，同时停掉老作业，将做好的savepoint用来触发新作业的执行。
>
>casel.chen 于2022年11月29日周二 08:38写道：
>
>> 线上有一个流量较大的flink sql作业需要升级添加业务字段，当前主要是kafka (canal) 多表关联写入 mongodb
>> 数据同步场景，除了source offset外无其他状态，如何让用户对升级无感呢？
>> 常规的停止作业再启动作业至少要几十秒，会造成消息积压告警，有没有可能先启新作业待运行平稳后再停
t;
> --原始邮件--
> 发件人:
> "user-zh"
>
> 发送时间:2022年11月30日(星期三) 晚上8:08
> 收件人:"user-zh"
> 主题:Re: flink sql作业无缝升级问题
>
>
>
> Fl
Flink目前无法做到无缝升级，需要走stop-with-savepoint、start
job的流程，但是在这之间可以做一些优化来缩短恢复时间。比如，把新作业先启动起来，申请好资源，同时停掉老作业，将做好的savepoint用来触发新作业的执行。
casel.chen 于2022年11月29日周二 08:38写道：
> 线上有一个流量较大的flink sql作业需要升级添加业务字段，当前主要是kafka (canal) 多表关联写入 mongodb
> 数据同步场景，除了source offset外无其他状态，如何让用户对升级无感呢？
> 常规的停止作业再启动作业
> On Nov 29, 2022, at 8:32 AM, casel.chen wrote:
>
> 业务需求是mysql订单表按天按供应商实时统计交易金额，订单表会发生修改和删除，用flink
> sql要如何实现呢？开窗取最新一条记录再聚合吗？如果遇到delete记录会不会减去相应的price呢？试着写了如下flink sql不知道对不对
会的，可以看下flink sql相关的原理文章，百度/谷歌一搜一大把。
祝好
Leonard
>
>
> select
> s.biddate,
>
我会选择 UDF + 配置文件，把配置文件放 HDFS上，UDF读这个配置文件。每次更新HDFS的配置文件，重启下任务
casel.chen 于2022年11月24日周四 12:01写道：
> 我有一个flink
> sql作业需要根据不同字段值满足不同条件来设置另一个字段值，还有一些嵌套if-else逻辑，这块逻辑不是固定的，业务方会过一段时间调整一次。
> 想请问如何使用flink sql优雅的处理嵌套if-else逻辑呢？我有想到使用drools规则引擎，通过udf来调用，不知道还有没有更好的办法？
>
>
flink sql有办法获取到rowkind元数据字段吗？比如按rowkind进行case when处理或者过滤
线上有一个流量较大的flink sql作业需要升级添加业务字段，当前主要是kafka (canal) 多表关联写入 mongodb
数据同步场景，除了source offset外无其他状态，如何让用户对升级无感呢？
常规的停止作业再启动作业至少要几十秒，会造成消息积压告警，有没有可能先启新作业待运行平稳后再停止老作业？kafka
group使用同一个，作业启动从group-offsets开始可以吗？另外，如果是有大状态作业又要如何无缝升级？
业务需求是mysql订单表按天按供应商实时统计交易金额，订单表会发生修改和删除，用flink
sql要如何实现呢？开窗取最新一条记录再聚合吗？如果遇到delete记录会不会减去相应的price呢？试着写了如下flink sql不知道对不对
select
s.biddate,
s.supplier,
sum(s.price)
from
(
select
*
from
(
select
biddate,
supplier,
price
hi
请问能在java代码里面try catch到基于flink sql写的任务异常信息吗？
双流关联场景下，流A数据到达后并不马上和流B进行关联，而是经过设置一段时间后再关联，这种场景下如何用flink
sql实现？如果当前不支持，需要怎样扩展flink sql呢？
我有一个flink sql作业需要根据不同字段值满足不同条件来设置另一个字段值，还有一些嵌套if-else逻辑，这块逻辑不是固定的，业务方会过一段时间调整一次。
想请问如何使用flink sql优雅的处理嵌套if-else逻辑呢？我有想到使用drools规则引擎，通过udf来调用，不知道还有没有更好的办法？
是的，类似阿里云和腾讯云上面的功能
在 2022-11-23 10:02:09，"Shengkai Fang" 写道：
>想问一下你想实现的功能是咋样的呢？是阿里云上的那种吗？
>
>Best,
>Shengkai
>
>casel.chen 于2022年11月23日周三 08:29写道：
>
>> flink sql社区版貌似还不支持CTAS/CDAS语句，请问要如何扩展flink
>> sql以支持CTAS/CDAS语句？可以给个思路或代码示例吗？谢谢！
想问一下你想实现的功能是咋样的呢？是阿里云上的那种吗？
Best,
Shengkai
casel.chen 于2022年11月23日周三 08:29写道：
> flink sql社区版貌似还不支持CTAS/CDAS语句，请问要如何扩展flink
> sql以支持CTAS/CDAS语句？可以给个思路或代码示例吗？谢谢！
flink sql社区版貌似还不支持CTAS/CDAS语句，请问要如何扩展flink sql以支持CTAS/CDAS语句？可以给个思路或代码示例吗？谢谢！
flink sql社区版貌似还不支持CTAS/CDAS语句，请问要如何扩展flink sql以支持CTAS/CDAS语句？可以给个思路或代码示例吗？谢谢！
可以了解下 TRY_CAST 是不是能满足需求.
https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/functions/systemfunctions/#type-conversion-functions
On Sat, Nov 19, 2022 at 4:31 PM kcz <573693...@qq.com.invalid> wrote:
> flink-1.16.0
> flink sql 类型转化遇到转化失败时候可以跳过这条数据嘛？
> 不想任务直接就挂了，
配置了prometheus收集flink sql作业指标，现在想根据这些指标动态设置一些告警规则，请问要如何实现？
查了下prometheus告警需要配置alert rule之后重启才生效，有没有办法不重启呢？常规实现方案是什么？
Hi,
Leonard 提到的 -i 参数可以满足你的需求。在初始化SQL文件中可以SET/RESET属性，CREATE/DROP等。
更多信息请查看：
https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/sqlclient/#sql-client-startup-options
Hi,
我记得有个-i 参数可以指定初始化sql文件，你贴你的初始化sql在文件里加进去就可以了。
祝好，
Leonard
> 2022年10月31日 下午4:52，casel.chen 写道：
>
> flink新版本已经找不到sql-clients-default.yaml文件了，那么之前配置的那些预置catalogs建议在哪里定义呢？通过初始化sql么？
flink新版本已经找不到sql-clients-default.yaml文件了，那么之前配置的那些预置catalogs建议在哪里定义呢？通过初始化sql么？
bSubmitter.java:240)
> at com.cgws.ccp.server.jobs.JobManager.submit(JobManager.java:189)
> at com.cgws.ccp.server.jobs.JobManager.submit(JobManager.java:156)
> at com.cgws.ccp.server.jobs
> .transitions.StartTransitionCallback.startJob(StartTransitionCallback.java:221)
SQL语句等周一上班时我再发出来
> 在 2022年10月22日，17:08，yinghua_zh 写道：
>
>
>
>> 在 2022年10月22日，17:06，yinghua_zh 写道：
>>
>> 感谢答复，Flink版本1.14.4
>>
> 在 2022年10月22日，17:01，Leonard Xu 写道：
>
> 你好，你的Flink 版本多少呀？我记得低版本才有这个问题。
> 另外SQL可以贴下嘛？
>
> 祝好，
> Leonard
>
>
>
> 在 2022年10月22日，17:06，yinghua_zh 写道：
>
> 感谢答复，Flink版本1.14.4
>
>>> 在 2022年10月22日，17:01，Leonard Xu 写道：
>>>
>>> 你好，你的Flink 版本多少呀？我记得低版本才有这个问题。
>>> 另外SQL可以贴下嘛？
>>>
>>> 祝好，
>>> Leonard
>>>
>>>
>>>
2022年10月22日 上午11:11，邮件帮助中心 写道：
>>>
>>> 大家好！
>>> 最近在开发一个项目时，在使用CDC表和维表表做Temporal Table
感谢答复，Flink版本14.4
> 在 2022年10月22日，17:01，Leonard Xu 写道：
>
> 你好，你的Flink 版本多少呀？我记得低版本才有这个问题。
> 另外SQL可以贴下嘛？
>
> 祝好，
> Leonard
>
>
>
>> 2022年10月22日 上午11:11，邮件帮助中心 写道：
>>
>> 大家好！
>> 最近在开发一个项目时，在使用CDC表和维表表做Temporal Table
>> JOIN时，发现2个表Join时join字段的类型必须一致，否则提交时提示如下的错误
>> The main method
你好，你的Flink 版本多少呀？我记得低版本才有这个问题。
另外SQL可以贴下嘛？
祝好，
Leonard
> 2022年10月22日 上午11:11，邮件帮助中心 写道：
>
> 大家好！
>最近在开发一个项目时，在使用CDC表和维表表做Temporal Table
> JOIN时，发现2个表Join时join字段的类型必须一致，否则提交时提示如下的错误
>The main method caused an error: Temporal table join requires an equality
> condition on fields of
大家好！
最近在开发一个项目时，在使用CDC表和维表表做Temporal Table
JOIN时，发现2个表Join时join字段的类型必须一致，否则提交时提示如下的错误
The main method caused an error: Temporal table join requires an equality
condition on fields of table.
为了解决上述问题，我们做了如下尝试：
1：在join时，对维表要关联的字段使用cast转换，如： JOIN ON CAST(tableA.filedA AS
"user-zh"
> <
> pens...@gmail.com;
> 发送时间:2022年10月14日(星期五) 中午11:55
> 收件人:"user-zh"
> 主题:Re: Flink SQL 中同时写入多个 sink 时，是否能够保证先后次序
>
>
>
> 好的，谢谢大家，之前也想过这个方案，复用/继承 JdbcDynami
ers,
>
> I have one question about watermark generating mechanism in Flink SQL.
> There are two mechanisms called *Periodic Watermarks* and *Punctuated
> Watermarks, *I want to use* Periodic Watermarks* with interval 5 seconds
> (meaning watermarks will be generated every 5 seconds), how s
Hi dear engineers,
I have one question about watermark generating mechanism in Flink SQL. There
are two mechanisms called Periodic Watermarks and Punctuated Watermarks, I want
to use Periodic Watermarks with interval 5 seconds (meaning watermarks will be
generated every 5 seconds), how
0月12日周三 16:59写道：
> >
> > Flink SQL 自身机制无法保证同一个作业多个 sink 的写入次序。 是否可以考虑从业务逻辑上动手脚，比如写入消息队列 sink 前加个
> udf
> > filter, udf 查询 database，满足条件才写入消息队列，当然这种方式对性能可能有影响。
> >
> > On Wed, Oct 12, 2022 at 2:41 PM Zhiwen Sun wrote:
> >
> > > hi all：
> > >
>
Flink SQL 自身机制无法保证同一个作业多个 sink 的写入次序。 是否可以考虑从业务逻辑上动手脚，比如写入消息队列 sink 前加个 udf
filter, udf 查询 database，满足条件才写入消息队列，当然这种方式对性能可能有影响。
On Wed, Oct 12, 2022 at 2:41 PM Zhiwen Sun wrote:
> hi all：
>
> 我们有个场景，需要 Flink SQL 同时写入消息和 database， 后续实时任务消费消息，再次读取 database， 如果消息先于
> database 写入，这就
Hi Zhiwen:
可以试试将kafkasink设置为exactly-once，同时下游任务的kafkasource设置为读已提交，不过如果你的checkpoint间隔很长，下游要很久才能从kafka拿到数据。
从 Windows 版邮件发送
发件人: Zhiwen Sun
发送时间: 2022年10月12日 14:42
收件人: user-zh@flink.apache.org
主题: Flink SQL 中同时写入多个 sink 时，是否能够保证先后次序
hi all：
我们有个场景，需要 Flink SQL 同时写入消息和 database， 后续实时任务消费
Hi
我理解应该是任务恢复的时候从上一次成功的 checkpoint 或者你指定的 checkpoint 里记录的 offset
开始消费，所以此时的统计值应该是有短暂的下跌，因为数据相当于回复到之前重复计算了一部分。这个应该是符合预期的，可能需要在业务上做一些处理。
Best
JasonLee
回复的原邮件
| 发件人 | 天下五帝东 |
| 发送日期 | 2022年10月10日 13:34 |
| 收件人 | user-zh@flink.apache.org |
| 主题 | Flink sql从ck恢复，统计数据波动问题 |
Hi:
各位大佬们
数据库的统计值
> 下面是被转发的邮件：
>
> 发件人: Hangxiang Yu
> 主题: 回复：Flink sql从ck恢复，统计数据波动问题
> 日期: 2022年10月10日 GMT+8 下午2:03:50
> 收件人: user-zh@flink.apache.org
> 回复－收件人: user-zh@flink.apache.org
>
> 是什么值下跌呢？哪个metric吗？
>
> On Mon, Oct 10, 2022 at 1:34 PM 天下五帝东 wrote:
>
>
ck“打”完是啥意思。
Congxian Qiu 于2022年10月10日周一 15:11写道：
>
> Hi
> 可以的话也同步下相关的计算逻辑，从 checkpoint 恢复后的统计结果可能会和计算逻辑有关
> Best,
> Congxian
>
>
> Hangxiang Yu 于2022年10月10日周一 14:04写道：
>
> > 是什么值下跌呢？哪个metric吗？
> >
> > On Mon, Oct 10, 2022 at 1:34 PM 天下五帝东 wrote:
> >
> > > Hi:
> > >
Hi
可以的话也同步下相关的计算逻辑，从 checkpoint 恢复后的统计结果可能会和计算逻辑有关
Best,
Congxian
Hangxiang Yu 于2022年10月10日周一 14:04写道：
> 是什么值下跌呢？哪个metric吗？
>
> On Mon, Oct 10, 2022 at 1:34 PM 天下五帝东 wrote:
>
> > Hi:
> > 各位大佬们，flink任务从状态恢复时，统计值有一个明显的下跌，等ck打完后，统计值才会上升，这种情况有什么好的解决方案吗？
>
>
>
> --
> Best,
> Hangxiang.
是什么值下跌呢？哪个metric吗？
On Mon, Oct 10, 2022 at 1:34 PM 天下五帝东 wrote:
> Hi:
> 各位大佬们，flink任务从状态恢复时，统计值有一个明显的下跌，等ck打完后，统计值才会上升，这种情况有什么好的解决方案吗？
--
Best,
Hangxiang.
Hi:
各位大佬们，flink任务从状态恢复时，统计值有一个明显的下跌，等ck打完后，统计值才会上升，这种情况有什么好的解决方案吗？
flink sql cdc 2.2.1消费mysql binlog遇到如下异常，有谁遇到过？发现作业自己做了重试后过去了，想知道异常的root
cause是什么？手动重起了作业重新消费后还是会出现。
Caused by:
com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.errors.ConnectException:
An exception occurred in the change event producer. This connector will be
stopped
Hi，
能麻烦问下你之前使用过哪个支持动态修改SQL的框架或系统吗？
Best，
Biao Geng
获取 Outlook for iOS<https://aka.ms/o0ukef>
发件人: Zeguang ZHANG
发送时间: Friday, September 30, 2022 3:26:22 PM
收件人: user-zh@flink.apache.org
主题: 提交新的flink-sql的最佳方案是什么？
您好，
我们在使用flink-13版本，使用flink-sql时候，每当更改s
您好，
我们在使用flink-13版本，使用flink-sql时候，每当更改sql，需要先cancel job然后 submit 新的 sql。
请问怎么做能在提交job时候不先取消job？
张泽光
Internal
目前还不支持 set HiveConf 里面的参数。
只能在创建 HiveCatalog 用的那个 hive-site.xml 里面改。
快发布的 Flink 1.16 就可以支持 set HiveConf 里面的参数。
Best regards,
Yuxia
- 原始邮件 -
发件人: "junjie miao"
收件人: "user-zh"
发送时间: 星期四, 2022年 9 月 22日 上午 10:19:06
主题: flink sql中如何对于hive进行hive的参数set
当使用flink sql inse
hi，when I write a sql like this:
String sqlCreate = "CREATE TABLE fs_table (\n" +
" `examplestr` bytes\n" +
") WITH (\n" +
" 'connector'='filesystem',\n" +
" 'format'='raw',\n" +
"
hi，when I write a sql like this:
String sqlCreate = "CREATE TABLE fs_table (\n" +
" `examplestr` bytes\n" +
") WITH (\n" +
" 'connector'='filesystem',\n" +
" 'format'='raw',\n" +
"
线上运行了多个flink sql作业，现在想监控端到端延迟。我配置了
metrics.latency.interval=3
metrics.latency.granularity=operator
metrics.latency.history-size=128
参数，延迟指标已经发到了prometheus，看到该指标有50、75、95、98，99，999分位线，另外还有operator_id和operator_id_subtask_index，细到了算子子task级别。
1. 想知道怎样根据这些暴露指标统计出该flink
sql作业的端到端延迟分位线？是需要把所有同一个job的
flink-sql-connector-hbase-2.2模块在shade打包时遗漏了commons-io依赖，导致当使用hbase
connector时开启lookup.async=true后，当执行结束时调用HBaseRowDataAsyncLookupFunction类的close()方法时调用asyncConnection.close()内部报错java.lang.ClassNotFoundException:
org.apache.flink.hbase.shaded.org.apache.commons.io.IOUtils。
Weihua
>
>
> On Fri, Aug 19, 2022 at 11:56 AM casel.chen wrote:
>
> > flink sql支持监听单个文件内容变化吗？文件中每一行是一条记录，对外输出的模式可以全量或者变量。
>
Hi，
不支持监听单个文件的变化，但是可以监听某个目录下文件的新增。
[1]
https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/filesystem/#source
Best,
Weihua
On Fri, Aug 19, 2022 at 11:56 AM casel.chen wrote:
> flink sql支持监听单个文件内容变化吗？文件中每一行是一条记录，对外输出的模式可以全量或者变量。
flink sql支持监听单个文件内容变化吗？文件中每一行是一条记录，对外输出的模式可以全量或者变量。
> 2022年7月22日 15:10，艾迪 <644860...@qq.com.INVALID> 写道：
>
> 我们目前使用flink
> sql输出数据到HDFS，存在多个sql从一个source消费并输出性能较差。通过反编译JVM中的代码发现，每个sql都将使用的到的source表的STRING类型字段深拷贝了一份。配置object-reuse后，没有改变。请问：
> flink sql 如何消除STRING类型字段的深拷贝
> 万分感谢
确定是深拷贝的问题吗？ 有拉火焰图看吗？ 具体的瓶颈是在哪个方法？
??flink ??1.12.3??
?? source table ?? kafka ??
source_table_kafka phoenix?? source_table_phoenix ,kafka join
phoenix count ?? sink ??kafka??count phoenix
??
select
t I need should be this one
> (UDTAGG):
> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/functions/udfs/#table-aggregate-functions
> As I want multiple rows as aggregate output. So my question: can we use
> UDTAGG in flink SQL?. If so, is there some gu
Hi,
Thanks for your response, I guess what I need should be this one (UDTAGG):
https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/functions/udfs/#table-aggregate-functions
As I want multiple rows as aggregate output. So my question: can we use UDTAGG
in flink SQL?. If so
question: does flink sql support UDTAGG? (user-defined table
aggregate function), seems only supported in flink table api? If not supported
in flink sql, how can I define an aggregated udf which could output multiple
rows to kafka.
Thanks for your help!
Regards,
Hunk
Hi dear engineers,
One small question: does flink sql support UDTAGG? (user-defined table
aggregate function), seems only supported in flink table api? If not supported
in flink sql, how can I define an aggregated udf which could output multiple
rows to kafka.
Thanks for your help
Hi，
flink sql如何写es的nested数组数据？
原始示例数据：
{
"id": "123",
"field1":[
{
"k1":1
},
{
"k1":1,
"k2":2
},
{
"k3":"3"
}
flink
sql??HDFS??sql??source??JVM??sql??sourceSTRING??object-reuse
flink sql STRING
Hi, 退订请发送任意内容至邮箱user-zh-unsubscr...@flink.apache.org
Best,
Zhanghao Chen
From: 孙福
Sent: Saturday, July 16, 2022 23:01
To: user-zh@flink.apache.org
Subject: Re:Re: Flink SQL 能处理延时丢弃的数据吗？
退订
在 2022-07-15 15:06:51，"Zhizhao Shangguan"
8605.html
> 的总结。不过这个特性还是实验性的，请谨慎使用。
>
>Best,
>Zhanghao Chen
>
>From: Zhizhao Shangguan
>Sent: Friday, July 15, 2022 10:44
>To: user-zh@flink.apache.org
>Subject: Flink SQL 能处理延时丢弃的数据吗？
>
>
Sent: Friday, July 15, 2022 10:44
To: user-zh@flink.apache.org
Subject: Flink SQL 能处理延时丢弃的数据吗？
Hi，ALL：
咨询个问题，对于超过watermark后的数据，还想继续触发窗口操作（类似API的allowedLateness机制），flink
SQL可以实现吗？ 可以话，要如何处理的？
Thanks♪(･ω･)ﾉ
你好，可以看下：https://www.mail-archive.com/issues@flink.apache.org/msg498605.html
的总结。不过这个特性还是实验性的，请谨慎使用。
Best,
Zhanghao Chen
From: Zhizhao Shangguan
Sent: Friday, July 15, 2022 10:44
To: user-zh@flink.apache.org
Subject: Flink SQL 能处理延时丢弃的数据吗？
Hi，ALL：
咨询个
Hi，ALL：
咨询个问题，对于超过watermark后的数据，还想继续触发窗口操作（类似API的allowedLateness机制），flink
SQL可以实现吗？ 可以话，要如何处理的？
Thanks♪(･ω･)ﾉ
在DataStream API中，part文件的名称可以通过OutputFileConfig来定义，文档详见[1]
而在Table
API中，FileSystemTableFactory中没有设计对应的Options，且在FileSystemOutputFormat.Builder中设置了默认的OutputFileConfig，prefix和partsuffix都为空字符串。
希望能够在FileSystemTableFactory中增加用以支持OutputFileConfig定义的Options，使得在flink
sql中可以完成精准控制FileSink输出的文件。
[1
共有 1922 项搜索結果，以下是第 101 - 200 matches
Mail list logo
