大佬们好,现在业务需求多,生产作业开始变得繁重,请教3个生产业务场景,主要是想看下各位有什么优雅的思路
1.kakfa table和group_ID应该怎么去维护呢?一个业务一张表吗?
2.如何做到复用表的效果?
3.如果新增一个业务需求,用到之前的一张kafka table,似乎要在一个程序里。执行多次 executeSql("sql
1")才不会乱序,但是这样的话,程序的耦合度会很高,请问该怎么优雅的处理这些场景呢?
--
Sent from: http://apache-flink.147419.n8.nabble.com/
这个应该和内存有关,我之前试过,存储的状态无限增长,导致运行几分钟后任务结束,并抛出异常,可以尝试一下加大内存和清理状态
| |
刘海
|
|
liuha...@163.com
|
签名由网易邮箱大师定制
在2021年1月4日 11:35,咿咿呀呀<201782...@qq.com> 写道:
我按照您这个修改了,跟我之前的也是一样的。能运行的通,输出的结果也是正确的,现在最大的问题是——运行一段时间后(3分钟左右)就出现了OSError:
Expected IPC message of type schema but got record batch这个错误
--
Sent
我按照您这个修改了,跟我之前的也是一样的。能运行的通,输出的结果也是正确的,现在最大的问题是——运行一段时间后(3分钟左右)就出现了OSError:
Expected IPC message of type schema but got record batch这个错误
--
Sent from: http://apache-flink.147419.n8.nabble.com/
Flink版本 1.12.0
发件人: amen...@163.com
发送时间: 2021-01-03 16:09
收件人: user-zh
主题: Flink SQL DDL Schema csv嵌套json
hi everyone,
zhangsan|man|28|goodst...@gmail.com|{"math":98, "language":{"english":89,
"french":95}}|china|beijing
这是一条来自kafka消息队列中的数据,当我创建kafka ddl为之定义schema时,报出异常信息:
Exception in
在程序自动重启恢复的时候,binlog可能被MySQL服务器删除了,导致debeziume connector读取了新的快照。
参考连接:https://debezium.io/documentation/reference/1.3/connectors/mysql.html#mysql-purges-binlog-files_debezium
--
Name:谢波
Mobile:13764228893
在 2021-01-04 10:38:30,"lingchanhu" 写道:
>sourcr:mysql-cdc
sourcr:mysql-cdc
sink:elasticsearch
问题描述:
从mysql中同步表数据至elasticsearch后,进行新增再删除的某条数据出现问题,导致sink失败(没加primary
key)。checkpoint失败,程序自动恢复重启后,checkpoint 成功,但是elasticsearch 中的数据是mysql
表中的两倍,出现重复同步情况。
程序的自动恢复不应该是从当前checkpoint 中记录的binlog 位置再同步么?为什么会再重头同步一次呢?
(ddl 中写死了server-id,
"
@Gordon Thanks a lot for the release and for being the release manager.
And thanks to everyone who made this release possible!
Best,
Xingbo
Till Rohrmann 于2021年1月3日周日 下午8:31写道:
> Great to hear! Thanks a lot to everyone who helped make this release
> possible.
>
> Cheers,
> Till
>
> On Sat, Jan
??Xingbo
hi!
java.io.FileNotFoundException: File file:/home/xjia/.flink/...
可以看出,从本地加载jar包,而不是hdfs。
我觉得可能是hadoop环境的问题,导致读取的scheme是file,使用 echo $HADOOP_CLASSPATH 检查你的环境。
Important Make sure that the HADOOP_CLASSPATH environment variable is set up
(it can be checked by running echo $HADOOP_CLASSPATH). If
Hi,
不好意思,这么晚才回复。因为你这个报错是发生在数据反序列的过程中,还没有到你写的函数体的具体内容。我看你pandas
udf的声明没有问题,那就得看下你的如何使用的了。我写了一个简化版本的,Array[String]作为输入,并且作为输出的,运行起来没有问题。
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings,
DataTypes
from pyflink.table.udf
社区的各位大神,有没有碰到这个问题的,请教。
--
Sent from: http://apache-flink.147419.n8.nabble.com/
谢谢你的回答。
但是我的col1,col2就已经是SMALLINT类型的了,我的问题是where条件中值下推过程中是Integer类型的,我希望值也是SMALLINT的。
祝好!
| |
automths
|
|
autom...@163.com
|
在2020年12月31日 18:17,whirly 写道:
Hi.
查询语句中可以使用 cast 内置函数将值强制转换为指定的类型,如 select CAST(A.`col1` AS SMALLINT) as col1
from table
参考:
Dear All,
Flink SQL>select * from table1;
在Flink客户端查询hive数据,结果每个字段全为NULL,但数据条数是对的,select
count是正确的,查具体数据就是NULL,不知何故?同样的查询在hive客户端是可以查到数据的
hive表时orc文件load的数据。
-
Thanks!
Jacob
--
Sent from: http://apache-flink.147419.n8.nabble.com/
??yarn-cluster23:50??/opt/module/hadoop3.2.1/bin/yarn
application -kill
application_1609656886263_0043??kill??job1:30??kill-job??state??
----
Great to hear! Thanks a lot to everyone who helped make this release
possible.
Cheers,
Till
On Sat, Jan 2, 2021 at 3:37 AM Tzu-Li (Gordon) Tai
wrote:
> The Apache Flink community released the second bugfix release of the
> Stateful Functions (StateFun) 2.2 series, version 2.2.2.
>
> *We
hi everyone,
zhangsan|man|28|goodst...@gmail.com|{"math":98, "language":{"english":89,
"french":95}}|china|beijing
这是一条来自kafka消息队列中的数据,当我创建kafka ddl为之定义schema时,报出异常信息:
Exception in thread "main" java.lang.IllegalArgumentException: Only simple
types are supported in the second level nesting of
17 matches
Mail list logo