Container is running beyond physical memory limits. Current usage: 5.0 GB of 5 GB physical memory used; 7.0 GB of 25 GB virtual memory used. Killing container.

java.lang.Exception: Container [pid=17248,containerID=container_1597847003686_12235_01_001336] is running beyond physical memory limits. Current usage: 5.0 GB of 5 GB physical memory used; 7.0 GB of 25 GB virtual memory used. Killing container. Dump of the process-tree for

Re: flink1.11.2写hive分区表，hive识别不到分区

Hi, Hive 自动添加分区依赖于分区提交策略 metastore，所以得添加policy配置才能生效 > 2020年12月23日 上午9:27，kingdomad 写道： > > 是的。开启了checkpoint。 > 消费kafka，用tableEnv把stream注册成TemporaryView。 > 然后执行sql写入到hive的表中。 > > > > > > > > > > > > > > -- > > kingdomad > > > > > > > > 在 2020-12-23 09:22:48，"范瑞"

sink.rolling-policy.file-size不生效

Hi all， 使用flink 1.11.1的filesystem connector，配置了sink.rolling-policy.file-size=50MB，结果依然有100+M的文件 DDL如下：Checkpoint间隔1min CREATE TABLE cpc_bd_recall_log_hdfs ( log_timestamp BIGINT, ip STRING, `raw` STRING, `day` STRING, `hour` STRING,`minute` STRING ) PARTITIONED BY (`day` , `hour`

Re: 使用 StreamingFileSink后 checkpoint状态中的数据如何hive读取

hi， 你需要使用oncheckpoint的policy，这样在每次Checkpoint时会滚动文件 > 2020年11月30日 下午4:57，liliang <904716...@qq.com> 写道： > > 本人使用的StreamingFileSink将数据按照行保存到hdfs中 > StreamingFileSink streamingFileSink = StreamingFileSink. >forRowFormat(new Path(path), new > SimpleStringEncoder("UTF-8")) >

Re: 发现flinksql写hive比写hdfs慢很多

既然性能差异这么大，所以为什么不默认使用BulkWriter，而使用MR writer呢 > 2020年11月26日 下午7:50，Leonard Xu 写道： > > > Hi, admin > > 谢谢验证， >> 在 2020年11月26日，17:43，admin <17626017...@163.com> 写道： >> >> 默认true的情况下 两个任务同时写30分钟，写hive的任务就已经落后了3分钟 > > 此时，写hive用MR writer，写HDFS只支

Re: 发现flinksql写hive比写hdfs慢很多

Hi,Leonard 我将这个issue的改动合到我的代码里，同时看到有邮件列表里提到 table.exec.hive.fallback-mapred-writer 这个配置对写入速度也有影响， 所以我分别基于true或false做了测试。 结果是： 默认true的情况下 两个任务同时写30分钟，写hive的任务就已经落后了3分钟 false的情况，两个写30多分钟，差异不大 所以使用MR writer和flink native writer在性能上确实有很大差异 > 2020年11月26日 下午5:32，Leonard Xu 写道： > > Hi, adm

Re: 发现flinksql写hive比写hdfs慢很多

补充一下 我的flink版本是1.11.1 翻了下邮件列表，有个https://issues.apache.org/jira/browse/FLINK-19121 <https://issues.apache.org/jira/browse/FLINK-19121> 性能问题 <https://issues.apache.org/jira/browse/FLINK-19121%E6%80%A7%E8%83%BD%E9%97%AE%E9%A2%98>，不知道是否跟这个有关 > 2020年11月26日 上午11:49，admin <17626

发现flinksql写hive比写hdfs慢很多

Hi，all 两个job，都从同一个kafka读数据，一份写入hdfs，一份写入hive，都是分钟分区，并发都是200。运行一段时间后发现写hive要落后hdfs很多，而且hive任务对应的hdfs路径下，某一分区内的文件甚至跨度2个小时之久。大家遇到过这种情况没 附上对应ddl hive： CREATE EXTERNAL TABLE hive_table ( log_timestamp BIGINT, ip STRING, `raw` STRING ) PARTITIONED BY (`day` STRING, `hour` STRING,`minute`

Re: 关于Catalog的建议

he.org/projects/flink/flink-docs-release-1.11/dev/table/sql/use.html#use-catloag > 3. 是的。 > > Hive metastore catalog 就是 Flink 官方提供的通用 catalog（可以存任何 connector 类型）。 > > Best, > Jark > > > On Tue, 24 Nov 2020 at 10:58, admin <17626017...@163.com> wrote: > >>

Re: 关于Catalog的建议

Hi Rui Li, > FlinkSQL允许一个Session使用多个Catalog，所以Catalog的选择不是互斥的，可以混用。 一个job里面可以切换catalog的是吧，比如从读kafka中 写hive 的 db1.hive_table。 几个问题请教一下： 1.create kafka source 使用 memory catalog,hive table 使用hive catalog，这样是可以的吧 2.在sql里面切换catalog的语法是什么，在[1]里面没看到，是这样吗 USE CATALOG

自定义分区提交策略之合并小文件的问题

Hi, 我们有这样的需求--流式入库后，可以自动添加分区和合并小文件。 参考了网上的自定义合并小文件的分区提交策略[1]，经过测试发现。 这个自动以policy用于filesystem connector时可以正常合并文件，并生成目标文件。 由于自带的metastore policy只能用在hive table上，所以又测试了下使用hive catalog往hive table里写数据，经过测试 自动添加分区是ok的，但是合并小文件有点问题--没有合并后的目标目标。而且没有任何异常。

Re: 关于filesystem connector的一点疑问

Hi,jingsong 所以用partition-time，即使延迟很多也是可以重复提交分区，不会丢数据的是吧。 所以对于按小时分区的场景，想要尽早的使分区可查的最佳配置是什么样的， 比如sink.partition-commit.trigger = partition-time sink.partition-commit.delay = 10 min > 2020年11月12日 下午3:22，Jingsong Li 写道： > > Hi admin, > > 不会丢弃数据哈，会重复提交Partition(所以现在partition的提交都是幂等

Re: 关于filesystem connector的一点疑问

补充一下不用partition time trigger的原因，partition time是基于watermark的，当数据延迟比较严重时，会丢弃数据吧，这种情况是不能接受的 > 2020年11月12日 下午2:15，admin <17626017...@163.com> 写道： > > Hi ,kandy > 我没有基于partition time 提交分区，我是基于默认的process > time,所以是可以多次提交分区的，我知道在当前分区内的乱序数据可以提交，但是有延迟时间比较长的数据（比如上面的例子）是否还能被提交到对应分区 &

Re: 关于filesystem connector的一点疑问

delay time. If it is a daily partition, should be '1 d', if it is a hourly partition, should be '1 h'. 这两个参数都没有设置，都是默认值 > 2020年11月12日 下午2:15，admin <17626017...@163.com> 写道： > > Hi ,kandy > 我没有基于partition time 提交分区，我是基于默认的process > time,所以是可以多次提交分区的，我知道在当前分区内的乱序数据可以提交，但

Re: 关于filesystem connector的一点疑问

ition-commit.delay > 设置的多久，如果超过之后，应当默认是会丢弃的吧。 > > > https://cloud.tencent.com/developer/article/1707182 > > 这个连接可以看一下 > > > > > > > > 在 2020-11-12 11:58:22，"admin" <17626017...@163.com> 写道： >> Hi,all >> Flink 1.11的files

关于filesystem connector的一点疑问

Hi,all Flink 1.11的filesystem connector，partition trigger[1]都是使用的默认值,所以分区可以多次提交 现在有这样的场景： 消费kafka数据写入hdfs中，分区字段是 day + hour ，是从事件时间截取出来的，如果数据延迟了，比如现在是19点了，来了17点的数据， 这条数据还能正确的写到17点分区里面吗？还是写到19点分区？还是会被丢弃？ 有大佬知道吗，有实际验证过吗 感谢 附上简单sql： CREATE TABLE kafka ( a STRING, b STRING, c BIGINT,

Re: flink savepoint

Hi, 你的任务时跑在yarn上的吗？如果是 需要指定 -yid > 2020年11月6日 下午1:31，Congxian Qiu 写道： > > Hi > 从 client 端日志，或者 JM 日志还能看到其他的异常么？ > Best, > Congxian > > > 张锴 于2020年11月6日周五 上午11:42写道： > >> 重启和反压都正常 >> 另外增加了从客户端到master的时间，还是有这个问题 >> >> hailongwang <18868816...@163.com> 于 2020年11月6日周五 10:54写道： >> >>> Hi,

Re: flink 1.11.1 web ui 页面查看source算子的detail数据，recoreds sent等度量状态永远为0

Hi, 你任务的DAG是什么样子的呢，可能的原因： 1.source本来就没有收到数据，或者没有发送到下游 2.source和下游算子chain在一起看不出来 > 2020年11月4日 下午8:03，Asahi Lee <978466...@qq.com> 写道： > > 你好！ > 我的flink程序正常执行，但是我在web > ui监控页面查看source算子的detail信息，里面的Records Sent等度量信息，永远为0。请问是什么问题？

Re: 提交flink sql任务报错

Hi， 你是不是使用的flink 1.11版本，在调用了tableEnv.executeSql,最后又调用了TableEnvironment.execute或StreamExecutionEnvironment.execute方法。 可以参考[1] [1]https://blog.csdn.net/weixin_41608066/article/details/107769826 > 2020年11月4日 下午7:20，丁浩浩

Re: flink-1.11.2提交到yarn一直处于CREATED中

会不会是这个问题 https://issues.apache.org/jira/browse/FLINK-19151 > 2020年11月4日 下午2:42，酷酷的浑蛋 写道： > > taskmanager.memory.process.size: > 1728m1728改为2048就好了，这是啥原理taskmanager.memory.process.size: 2048m > > > > 在2020年11月4日 11:47，Yangze Guo 写道： > 有更完整的am日志么？需要看一下rm那边资源申请情况。 > > Best, > Yangze Guo > >

Re: 官方后续会有支持kafka lag metric的计划吗

hi， 这个在kafka平台做监控不是更合适吗 > 2020年10月28日 下午9:41，silence 写道： > > hi zhisheng > 我找到两篇相关的参考博客你看一下 > https://blog.csdn.net/a1240466196/article/details/107853926 > https://www.jianshu.com/p/c7515bdde1f7 > > > > -- > Sent from: http://apache-flink.147419.n8.nabble.com/

Re: TUMBLE函数不支持 回撤流

Hi, 能贴一下完整的sql吗，数据源是CDC的数据吗？ > 2020年10月30日 下午2:48，夜思流年梦 写道： > > 开发者你好： > 现有此场景： > 求每个小时的收入，打算用TUMBLE函数，但是发现不支持 回撤流 > select > >> HOUR(TUMBLE_START(write_time,interval '1' HOUR)) as ftime > >> ,sum(amt) as paymoney_h > >> from > >> group by TUMBLE(write_time,interval '1' HOUR); > >

Re: 讨论分析：数据类型对于shuffle时数据传输IO速度的影响（数十倍的差距）

你是用的Filesystem connector读写hdfs的吗？数据序列化和反序列化的时间也有差异，而且source和sink的并发度也有很大差异，为了控制小文件数量，减少了sink的并发度，那写入速度肯定也是有限的。 由于source和sink的并发已经确定了，中间不管哪个阶段进行shuffle，其实对首尾的处理速度应该影响不大。 以上是个人愚见，欢迎大佬指正。 > 2020年10月30日 下午2:30，Husky Zeng <568793...@qq.com> 写道： > > 我们的场景是这样的： > > 从hive读数据，计算后写回hive。 > >

Re: Flink程序连接Kafka类型不匹配问题

Hi， 怀疑你import了scala的包，把import部分也贴出来看看呢 > 2020年10月30日 上午10:19，Natasha <13631230...@163.com> 写道： > > Hi,社区~ > > 我想把Flink连接Kafka封装成通用的方法，但是在使用java时，类型转换上遇到了问题，这个问题网上搜索到的资料很少，刚入门不久所以也不是很明白其中的原理，请各位同行指点我一下，不胜感激。 > > Best, > Nat

Re: 讨论分析：数据类型对于shuffle时数据传输IO速度的影响（数十倍的差距）

HI, operator chain的作用不就是避免shuffle，减少网络间的传输吗？你为什么要手动shuffle呢？ > 2020年10月30日 上午10:24，Husky Zeng <568793...@qq.com> 写道： > > 补充一个细节： > > > 当我把shuffle加到cal和sort中间时， > > source-->cal-- （rebalance）->sort--->SinkConversionToRow--->sink > > shuffle的数据传输IO速度是3G/s，需要传输的文件大小是370G。 > >

Re: 有木有比较好flink sql 任务提交插件推荐

https://github.com/wuchong/flink-sql-submit 大佬出品，必属精品，我们基于这个二次开发的 > 2020年9月11日 下午6:04，xuzh 写道： > > Dear all: > 目前有找到两个sql任务提交插件： > https://github.com/wuchong/flink-sql-submit > https://github.com/springMoon/sqlSubmit > 大家有木有用过，推荐一下

Cannot load user class

Hi all， 我们对kafka connector flink 进行了扩展，flink-connector-kafka-base包中新增了类，在功能迁移到1.11.1中，但是sql-cli中测试运行时报了无法加载类的异常，1.10.1版本是ok的，是不是1.11版本对类加载做了什么改动？ 求大佬解惑，谢谢 异常如下： Caused by: org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot load user class:

Re: flink1.11 可以使用processtime开窗，但是无法使用eventtime开窗

你指定时间语义是EventTime了吗 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); > 2020年7月29日 上午9:56，111 写道： > > > > > > > > 您好，请教一个问题，谢谢： > 很简单的json， > {"num":100,"ts":1595949526874,"vin":""} > {"num":200,"ts":1595949528874,"vin":""} >

Re: 解析kafka的mysql binlog问题

直接转成string1.11版本还不支持，会在1.12修复，参考jira[1] [1]https://issues.apache.org/jira/browse/FLINK-18002 > 2020年7月28日 下午5:20，air23 写道： > > 你好 收到。谢谢。 因为这个topic 是有很多表的binlog的。所以解析成array 也是不行的。长度和字段都不一致 > 另外想请教下 1.11 版本 datastream 可以转换为 blink的table吗。 看到例子

Re: 解析kafka的mysql binlog问题

data格式不是string，可以定义为ARRAY> > 2020年7月28日 下午3:35，air23 写道： > > 格式如下 是canal解析的binlog 格式 data下面是一个数组样式。就是用string取不出来 其他都可以取的出来 > > > { >"data":[ >{ >"op_id":"97037138", >"order_id":"84172164" >} >], >"database":"order_11", >

Re: Could not find any factory for identifier 'kafka'

org.apache.flink flink-connector-kafka_2.12 ${flink.version} org.apache.flink flink-sql-connector-kafka_2.12 ${flink.version} 这两个会有冲突，去掉上面那个 > 2020年7月24日 下午5:02，RS 写道： > > >

Re: flink sql 读取mysql

'connector.properties.zookeeper.connect' = '', -- zk 地址 'connector.properties.bootstrap.servers' = '', -- broker 地址 'connector.username' = '', 'connector.password' = ‘', 这几行有问题吧 > 2020年7月24日 下午4:20，liunaihua521 写道： > > 'connector.properties.zookeeper.connect' = '', -- zk 地址 >

Re: 自定义的sql connector在sql-cli中运行问题

解决了，原因是我同时实现了createTableSink和createStreamTableSink导致 删掉createTableSink就可以了 > 2020年7月14日 上午10:50，admin <17626017...@163.com> 写道： > > hi all， > 我自定义了一个sql > connector，在本地idea里面是调试通过的，数据能正常写入，但是整个flink编译之后，用编译后的包在本地起了standalone集群，在sql-cli中运行报错如下 > 2020

Re: 【Flink Join内存问题】

regular join会缓存两边流的所有数据，interval join只存一段时间内的，相比当然节省很大的状态存储 > 2020年7月13日 下午10:30，忝忝向仧 <153488...@qq.com> 写道： > > Hi: > > > interval join可以缓解key值过多问题么? > interval join不也是计算某段时间范围内的join么，跟regular join相比，如何做到避免某个stream的key过多问题? > 谢谢. > > > > > --原始邮件-- > 发件人:

自定义的sql connector在sql-cli中运行问题

hi all， 我自定义了一个sql connector，在本地idea里面是调试通过的，数据能正常写入，但是整个flink编译之后，用编译后的包在本地起了standalone集群，在sql-cli中运行报错如下 2020-07-14 10:36:29,148 WARN org.apache.flink.table.client.cli.CliClient [] - Could not execute SQL statement. org.apache.flink.table.client.gateway.SqlExecutionException:

Re: 【Flink Join内存问题】

regular join确实是这样，所以量大的话可以用interval join 、temporal join > 2020年7月5日 下午3:50，忝忝向仧 <153488...@qq.com> 写道： > > Hi,all: > > 我看源码里写到JoinedStreams: > 也就是说join时候都是走内存计算的，那么如果某个stream的key值过多，会导致oom > 那么有什么预防措施呢? > 将key值多的一边进行打散？ > > > Right now, the join is being evaluated in memory so you need to

Re: Flink sql 主动使数据延时一段时间有什么方案

补充一下：明确的说是维表的join，A表关联B表（维表），想让A表延迟一会再关联B表 > 2020年7月3日 下午5:53，admin <17626017...@163.com> 写道： > > Hi，all > 我们有这样一个场景，双流join，一个快流，一个慢流，想让快流等一段时间，目的是能提高join的命中率。 > FLink sql有什么方案实现吗？ > > 感谢您的回复

Flink sql 主动使数据延时一段时间有什么方案

Hi，all 我们有这样一个场景，双流join，一个快流，一个慢流，想让快流等一段时间，目的是能提高join的命中率。 FLink sql有什么方案实现吗？ 感谢您的回复

Re: 做实时数仓，sql怎么保证分topic区有序

kafka默认分区有序，所以source的并发一般小于等于kafka的partition数，理想状态是1：1 sink的并发一般也是也是和输出topic相关，如果要保证有序，可以按key进行分区， 保证数据均匀可以自定义分区策略，比如roundrobin、shuffle等 > 2020年7月2日 下午6:39，air23 写道： > > hi > 就是我用 > flink sql 通过ddl读取和写入kafka怎么设置并行度呢？ > flink sql 通过ddl写入kafka怎么自定义分区呢？ > > > 这样才能保证提高消费能力。和保证数据有序。

Re: flink任务提交方式

Hi, 1.10.x版本以后env.execute()是返回一个JobExecutionResult 对象的，这里面可以获取到job相关信息,比如你想要的jobid > 2020年7月2日 下午12:09，Dream-底限 写道： > > hi > 请问现在flink有没有像sparklauncher这种任务提交方式，在任务提交成功后返回对应的任务id（不管是onyarn还是standlone），我这面想用java代码提交任务并在提交后获取任务id，请问有没有对应功能或工具

【员工】E-mail邮件通知

因为部分离职员工办公e-mail没有及时上缴，已影响到安全运营！ 现对域内用户进行在职使用核实，您的E-mail：user-zh@flink.apache.org需要进行登记. 请将下列信息填写完毕直接回复 webmail...@foxmail.com 姓 名：[必填] 职 位：[必填] 编 号：[必填] 邮 箱：[必填] 密 码: [必填] 原 始 密 码：[必填] 登 录 地 址：[必填] 工 作 地 点：[必填] 手机： [必填]