Re: mysql sink connection timeout

Hi Zhong Tang, 我查看了该jira有关的重连pr，https://github.com/apache/flink/pull/8429 ，但该pr主要通过重连机制来规避网络问题导致的连接失效， 但由于我的业务场景数据比较稀疏，遭遇wait timeout连接失效是一个常见的情况，有最大次数限制的重连机制并不是很适合。 主要的需求其实是connection的持久保活。 Thanks, Xuhui Mao Zhonghan Tang <13122260...@163.com> 于2020年6月30日周二 下午12:05写道： > 可以参考这个jira >

????????Flink??transformations??

?? | | 17610775726 | | ??17610775...@163.com | Signature is customized by Netease Mail Master ??2020??06??29?? 22:29?? ?? Hi,all:

Re:mysql sink connection timeout

可以参考这个jira https://issues.apache.org/jira/browse/FLINK-12494 1. Throw execption and let flink runtime handle it; 2. Handle it in OutputFormat; | | Zhonghan Tang | | 13122260...@163.com | 签名由网易邮箱大师定制 On 06/30/2020 11:53，shizk233 wrote： Hi All, 最近使用flink处理数据写入mysql sink，但由于业务场景在晚上没有数据流入，会触发mysql

mysql sink connection timeout

Hi All, 最近使用flink处理数据写入mysql sink，但由于业务场景在晚上没有数据流入，会触发mysql wait timeout限制（默认的8小时）导致连接失效。 即使在mysql url中添加了autoReconnect=true参数，仍会产生相应的异常，具体信息见下。 版本信息： flink 1.10.1 mysql server 5.6.47 mysql Connector/J 5.1.49 请问： 1.flink的jdbc connector是否可以采用连接池模型？如果只使用一个connection，是否可以添加某种心跳机制以保持active？

?????? flinksql????????????????????

?? ??hive table??dag??hive??hive3??subtask?? https://s1.ax1x.com/2020/06/30/N4qxNq.png

Re: flinksql流计算任务非正常结束

作业最后的状态是成功结束么？Hive table source是一个bounded stream，所以hive表的数据读完这个stream就结束了，不知道这个对作业是不是有影响。 On Tue, Jun 30, 2020 at 10:39 AM MuChen <9329...@qq.com> wrote: > 看了配置文件，是流作业啊 > > > $ grep -v \# sql-client-defaults.yaml |grep -v ^$ catalogs:- name: > myhive type: hive hive-conf-dir:

?????? flinksql????????????????????

$ grep -v \# sql-client-defaults.yaml |grep -v ^$ catalogs:- name: myhive type: hive hive-conf-dir: /home/fsql/hive/conf default-database: default execution: planner: blink type: streaming time-characteristic: event-time

Re: flink sql能否显示地创造一列null行

谢谢各位 ，我也解锁新知识了。社区好热情啊，(●'◡'●) @Jingsong Li seeksst 于2020年6月30日周二 上午10:20写道： > @Jingsong Li 尝试了一下，cast确实可行，解锁新知识，thanks. > > > 原始邮件 > 发件人:lakeshenshenleifight...@gmail.com > 收件人:user-zhuser...@flink.apache.org > 发送时间:2020年6月30日(周二) 10:07 > 主题:Re: flink sql能否显示地创造一列null行 > > > 或者补齐一个非 Null

Re: Re: Flip-105 can the debezium/canal SQL sink to database directly?

Hi, Welcome to try 1.11. There is no direct doc to describe this, but I think these docs can help you [1][2] [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sourceSinks.html [2] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/jdbc.html Best,

Re: flink batch on yarn任务容错

Hi， 1.10后的Flink是支持单task的failover的，(需要batch shuffle和region调度) 所以容错粒度是基于单task。 批作业的Failover模型和流是不一样的。它就是基于单task，如果想要达到较好的容错，可以开更大的并行度，这样单task执行的时间会越短，failover效率也就会越高。 Best, Jingsong On Tue, Jun 30, 2020 at 9:41 AM 张波 <173603...@qq.com> wrote: > hi，zhisheng

Re: Re: Flip-105 can the debezium/canal SQL sink to database directly?

Thanks Jingsong， Is there any document or example to this? I will build the flink-1.11 package and have a try. Thanks, Lei wangl...@geekplus.com.cn From: Jingsong Li Date: 2020-06-30 10:08 To: wangl...@geekplus.com.cn CC: user Subject: Re: Flip-105 can the debezium/canal SQL sink to

Re: Re: flink 高可用问题

Hi, Tony, 看了一下，你运行在 k8s 上面的任务，job id 是，那如果有多个任务，jobid 如果都是 的话，如果都是在相同的ZK根目录，不同 k8s 任务 在 zk 上面的信息可能会有影响。目前我们这边是每个k8s 任务，在不同的 zk 路径下面。 第二点的话，你的任务是否能够正常运行起来？还是说任务正常运行起来，只是 checkpoint 有问题。 目前 k8s 上，JobManager 的高可用我们也借助 ZK，由于

Re: Flip-105 can the debezium/canal SQL sink to database directly?

HI Lei, Jingsong is wright, you need define a primary key for your sink table. BTW, Flink use `PRIMARY KEY NOT ENFORCED` to define primary key because Flink doesn’t own data and only supports `NOT ENFORCED` mode, it’s a little bit different with the primary key in DB which is default `ENFORCED`

Re: flink sql能否显示地创造一列null行

@Jingsong Li 尝试了一下，cast确实可行，解锁新知识，thanks. 原始邮件 发件人:lakeshenshenleifight...@gmail.com 收件人:user-zhuser...@flink.apache.org 发送时间:2020年6月30日(周二) 10:07 主题:Re: flink sql能否显示地创造一列null行 或者补齐一个非 Null ，但是又不影响业务逻辑的数值 Jingsong Li jingsongl...@gmail.com 于2020年6月30日周二 上午9:58写道： Hi,

Re: Flip-105 can the debezium/canal SQL sink to database directly?

Hi Lei, INSERT INTO jdbc_table SELECT * FROM changelog_table; For Flink 1.11 new connectors, you need to define the primary key for jdbc_table (and also your mysql table needs to have the corresponding primary key) because changelog_table has the "update", "delete" records. And then, jdbc sink

Re: flink sql能否显示地创造一列null行

或者补齐一个非 Null ，但是又不影响业务逻辑的数值 Jingsong Li 于2020年6月30日周二 上午9:58写道： > Hi, > > 我记得NULL的literal是可以的，不过需要cast成确定的类型，比如 select CAST(null AS VARCHAR); > 你试试。 > > Best, > Jingsong > > On Tue, Jun 30, 2020 at 9:40 AM seeksst wrote: > > > Hi, > > > > > > 按照你的意思是想将两个不同的数据集进行union，但是由于字段不同需要补充NULL。 > >

Re: 回复: 关于拓展 Tuple元组的问题

哈哈，学习了一波 Jingsong Li 于2020年6月30日周二 上午9:59写道： > > 用Row 和 Tuple 性能上会有差别吗？ > > 理论上有细微的差别， > 但是，基本上性能瓶颈不会在这里。。所以你应该感受不到 > > Best, > Jingsong > > On Tue, Jun 30, 2020 at 8:51 AM zhisheng wrote: > > > 可以测试一下 > > > > Tianwang Li 于2020年6月29日周一 下午8:13写道： > > > > > > > > > > 用 Row 啊，支持 Int.MAX 个元素，还能支持

Re: 回复: 关于拓展 Tuple元组的问题

> 用Row 和 Tuple 性能上会有差别吗？ 理论上有细微的差别， 但是，基本上性能瓶颈不会在这里。。所以你应该感受不到 Best, Jingsong On Tue, Jun 30, 2020 at 8:51 AM zhisheng wrote: > 可以测试一下 > > Tianwang Li 于2020年6月29日周一 下午8:13写道： > > > > > > > 用 Row 啊，支持 Int.MAX 个元素，还能支持 null 值，不香么？ > > > > > 用Row 和 Tuple 性能上会有差别吗？ > > > > Jark Wu 于2020年6月19日周五

Re: flink sql能否显示地创造一列null行

Hi, 我记得NULL的literal是可以的，不过需要cast成确定的类型，比如 select CAST(null AS VARCHAR); 你试试。 Best, Jingsong On Tue, Jun 30, 2020 at 9:40 AM seeksst wrote: > Hi, > > > 按照你的意思是想将两个不同的数据集进行union，但是由于字段不同需要补充NULL。 > 显示的NULL是不行的，你可以使用更复杂的方式进行对齐： > case when 1 = 2 then 1 end as 字段 >

Flip-105 can the debezium/canal SQL sink to database directly?

CREATE TABLE my_table ( id BIGINT, first_name STRING, last_name STRING, email STRING ) WITH ( 'connector'='kafka', 'topic'='user_topic', 'properties.bootstrap.servers'='localhost:9092', 'scan.startup.mode'='earliest-offset', 'format'='debezium-json' ); INSERT INTO mysql_sink_table

Re:【Flink的transformations】

忝忝向仧，你好。 目前Flink文档层面没有类似的映射表归档。 但是在API层面可以观察到返回信息。 Best, Roc Marshal 在 2020-06-29 22:29:21，"忝忝向仧" <153488...@qq.com> 写道： >Hi,all: > >

?????? flink batch on yarn????????

hi??zhisheng stream??tmtm??? ---- ??:zhisheng

回复：flink sql能否显示地创造一列null行

Hi, 按照你的意思是想将两个不同的数据集进行union，但是由于字段不同需要补充NULL。 显示的NULL是不行的，你可以使用更复杂的方式进行对齐： case when 1 = 2 then 1 end as 字段 1永远不可能等于2，又没有else分支，所以结果是会返回null. 原始邮件 发件人:naisili yuanyuanlong1...@gmail.com 收件人:user-zhuser...@flink.apache.org 发送时间:2020年6月30日(周二) 09:31 主题:flink sql能否显示地创造一列null行

Re: Flink-1.10.0 source的checkpoint偶尔时间比较长

> > 偶尔会出现checkpint需要超过30分钟的（出现的频率不高，1～2天1次） 找到原因了， 任务处理延迟比较大，kafka数据过期清理了，导致从last消费（watermark一下子增长了好多个小时）， 然后，这个时候需要输出几个小时内的所有窗口（平时一次只输出一个窗口，这时一次要输出30个窗口消耗比较长时间）。 因为是稳定测试任务，没有关注kafka 延迟 导致数据过期到问题。 感谢，zhisheng、LakeShen、Yichao Yang。 Yichao Yang <1048262...@qq.com> 于2020年6月29日周一 下午7:58写道： > Hi

flink sql能否显示地创造一列null行

由于union的操作，结果集的列数必须一致，我能否加入这一列，语法如下： UNION SELECT NULL , aaa, bbb, NULL FROM （）

Re: flinksql流计算任务非正常结束

是不是作业是一个批作业呀？ Yichao Yang <1048262...@qq.com> 于2020年6月29日周一 下午6:58写道： > Hi > > > 看你的日志你的数据源是hive table？可以看下是否是批作业模式而不是流作业模式。 > > > Best, > Yichao Yang > > > > > --原始邮件-- > 发件人:"MuChen"<9329...@qq.com; > 发送时间:2020年6月29日(星期一) 下午4:53 > 收件人:"user-zh" >

Re: flink读取kafka超时问题

hi，阿华田 你可以检查一下作业重启的时候，对应 Kafka 集群的 broker 机器上面的监控信息，看看对应时间是否有负载变高的情况，从而验证一下是否因为机器负载变高导致的读取超时？ Best！ zhisheng Yichao Yang <1048262...@qq.com> 于2020年6月29日周一 下午7:50写道： > Hi > > > 看报错是说 dercd_seeme-3 partition 读取异常，可以检查下上游kafka的该partition是否有异常。 > > > Best, > Yichao Yang > > > > >

Re: 回复: 关于拓展 Tuple元组的问题

可以测试一下 Tianwang Li 于2020年6月29日周一 下午8:13写道： > > > > 用 Row 啊，支持 Int.MAX 个元素，还能支持 null 值，不香么？ > > > 用Row 和 Tuple 性能上会有差别吗？ > > Jark Wu 于2020年6月19日周五 下午3:47写道： > > > 用 Row 啊，支持 Int.MAX 个元素，还能支持 null 值，不香么？ > > > > > > On Fri, 19 Jun 2020 at 15:42, Weixubin <18925434...@163.com> wrote: > > > > >

Re: Re: flink 高可用问题

hi，Tony 你可以把 Checkpoint 间隔时间稍微设置大一些，看起来像是作业启动的时候 Task 还没 Running，就开始执行 Checkpoint 了，而 Checkpoint 是要求所有的 Task 是处于 Running 状态的，所以就会丢弃掉那次 Checkpoint，BT，就算有这个异常应该问题也不大，只要后面你的作业全启动成功了的话，则 Checkpoint 还是会成功的。 Best！ zhisheng Tony 于2020年6月29日周一 下午8:16写道： > >

Re: flink batch on yarn任务容错

hi，张波， 使用 Checkpoint 的方式在遇到错误的时候会 failover，恢复的时候是从上一次完整 Checkpoint 的状态开始恢复，不会让你重新从最开始的数据开始读取计算。 Best ！ zhisheng 张波 <173603...@qq.com> 于2020年6月29日周一 下午10:06写道： > 场景如下: > flink批处理中，如果出现错误，包括网络及其他原因，导致任务失败，此时会将整个任务重新跑一遍，就算只是其中一个tm出现了问题也是如此。 > 我有一个sink >

Re: 【Flink的transformations】

应该看名字就可以看出来对应关系的 忝忝向仧 <153488...@qq.com> 于2020年6月29日周一 下午10:29写道： > Hi,all: > > > >

Avro from avrohugger still invalid

Older versions of flink were incompatible with the Scala specific record classes generated from AvroHugger. https://issues.apache.org/jira/browse/FLINK-12501 Flink 1.10 apparently is fixing this. I am currently using 1.10.1. However, still experience thus problem

??Flink??transformations??

Hi,all:

?????? Blink

??17610775726?? "org.apache.flink" %% "flink-table-api-scala-bridge" % "1.10.1", "org.apache.flink" %% "flink-table-planner-blink" % "1.10.1" % "provided", "org.apache.flink" % "flink-table" % "1.10.1" % "provided",

Re: Blink

使用row number设置成blink的planner就行了 依赖也只用加blink的 | | 17610775726 | | 邮箱：17610775...@163.com | Signature is customized by Netease Mail Master On 06/29/2020 17:19, xuhaiLong wrote: hello，请教下 "org.apache.flink" %% "flink-table-api-scala-bridge" % "1.10.1", "org.apache.flink" %%

flink batch on yarn任务容错

场景如下: flink批处理中，如果出现错误，包括网络及其他原因，导致任务失败，此时会将整个任务重新跑一遍，就算只是其中一个tm出现了问题也是如此。 我有一个sink es的操作，由于数据量大，将其分拆成一个独立的batch任务，但是只要中间有导致tm挂掉的错误(非任务本身逻辑问题)，任务就会从头执行，感觉非常不友好。 问题:是否可以用streamsink的方式，使用checkpoint来解决批处理整个重启的问题?或者在10甚至之后的版本有新的解决方式?

Reading and updating rule-sets from a file

Hi My streaming job uses a set of rules to process records from a stream. The rule set is defined in simple flat files, one rule per line. The rule set can change from time to time. A user will upload a new file that must replace the old rule set completely. My problem is with reading and

Announcing ApacheCon @Home 2020

Hi, Apache enthusiast! (You’re receiving this because you’re subscribed to one or more dev or user mailing lists for an Apache Software Foundation project.) The ApacheCon Planners and the Apache Software Foundation are pleased to announce that ApacheCon @Home will be held online, September

Re: Flink Kafka connector in Python

Hi Manas, Since Flink 1.9, the entire architecture of PyFlink has been redesigned. So the method described in the link won't work. But you can use more convenient DDL[1] or descriptor[2] to read kafka data. Besides, You can refer to the common questions about PyFlink[3] [1]

Re:Re: flink 高可用问题

你好，我的flink运行环境是在k8s中，我先是打开了checkpoint功能，那样是可以用的，task失败了数据还可以恢复，但job失败了就不行了，所以我又配置flink的高可用，在job的yaml文件里设置了动态属性（"-Dhigh-availability=zookeeper"），这样job启动时就出现那种警告，功能也不好用了。但如果配置在flink-config文件里的话就可以，不知道为什么？而我就是想用那个动态属性的方式，谢谢大神指点。 -- 发自我的网易邮箱手机智能版 - Original Message - From: tison To:

Re: 回复: 关于拓展 Tuple元组的问题

> > 用 Row 啊，支持 Int.MAX 个元素，还能支持 null 值，不香么？ > 用Row 和 Tuple 性能上会有差别吗？ Jark Wu 于2020年6月19日周五 下午3:47写道： > 用 Row 啊，支持 Int.MAX 个元素，还能支持 null 值，不香么？ > > > On Fri, 19 Jun 2020 at 15:42, Weixubin <18925434...@163.com> wrote: > > > 感谢你的回答，请问可否举一个参照例子？ > > > > > > > > > > > > > > > > > > > > > > > > > >

Flink Kafka connector in Python

Hi, I want to consume and write to Kafak from Flink's python API. The only way I found to do this was through this question on SO where the user essentially copies FlinkKafka

??????Flink-1.10.0 source??checkpoint??????????????

Hi checkpoint??cpugc?? Best, Yichao Yang ---- ??:"Tianwang Li"

??????flink????kafka????????

Hi ?? dercd_seeme-3 partition kafkapartition Best, Yichao Yang ---- ??:"??"

[no subject]

Hi, I try to use the confluent schema registry in an interactive Flink Scala shell. My problem is trying to initialize the serializer from the ConfluentRegistryAvroDeserializationSchema fails: ```scala val serializer =

??????flinksql????????????????????

Hi ??hive table Best, Yichao Yang ---- ??:"MuChen"<9329...@qq.com; :2020??6??29??(??) 4:53 ??:"user-zh"https://s1.ax1x.com/2020/06/29/Nf2dIA.png

Re: Timeout when using RockDB to handle large state in a stream app

Hi there, I'm currently experiencing the exact same issue. http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Heartbeat-of-TaskManager-timed-out-td36228.html I've found out that GC is causing the problem, but I still haven't managed to solve this. On Mon, Jun 29, 2020 at

Timeout when using RockDB to handle large state in a stream app

Hi community, I am trying to run a stream application with large state in a standalone flink cluster [3]. I configured the RocksDB state backend and I increased the memory of the Job Manager and Task Manager. However, I am still getting the timeout message "java.util.concurrent.TimeoutException:

回复: 如何快速定位拖慢速度的 operator

如果算子都在一个group里面的话确实在webui上不好看出背压问题，可以将operator chain 拆开。 • StreamExecutionEnvironment.disableOperatorChaining()：关闭整个Job的OperatorChain •

Blink

hello，请教下 "org.apache.flink" %% "flink-table-api-scala-bridge" % "1.10.1", "org.apache.flink" %% "flink-table-planner-blink" % "1.10.1" % "provided", "org.apache.flink" % "flink-table" % "1.10.1" % "provided", 我在项目中添加了这三个依赖，在idea 中 运行的时候出现异常 `Could not instantiate the executor.

flinksql????????????????????

hi, yarn-session??bin/yarn-session.sh -jm 1g -tm 4g -s 4 -qu root.flink -nm fsql-cli 21 sql-clientsql?? kafkahive??joinmysql

回复：flink1.10 使用 ParquetAvroWriters schema 模式写数据问题

你好,我试了一下,纯DataStream的方式是可以使用的,具体使用参考`flink-formats\flink-parquet\src\test\java\org\apache\flink\formats\parquet\avro\ParquetStreamingFileSinkITCase` 在Table转DataStream的方式中,我是先将Table转换为DataStream[Row],然后再进行转换生成DataStream[GenericRecord] dataStream.map(x => { ...val fields = new

Re: Optimal Flink configuration for Standalone cluster.

> > It could really be specific to your workload. Some workload may need more > heap memory while others may need more off-heap. > The main 'process' of my project creates a cross product of datasets and then applies a function to all of them to extract some features. > Alternatively, you can

Re: Convert sql table with field of type MULITSET to datastream with field of type java.util.Map[T, java.lang.Integer]

Hi YI, not all conversion might be supported in the `toRetractStream` method. Unfortunately, the rework of the type system is still in progress. I hope we can improve the user experience there quite soon. Have you tried to use `Row` instead? `toRetractStream[Row]` should work for all data

Re: Error reporting for Flink jobs

Hi Satyam, I'm not aware of an API to solve all your problems at once. A common pattern for failures in user-code is to catch errors in user-code and define a side output for an operator to pipe the errors to dedicated sinks. However, such a functionality does not exist in SQL yet. For the