用,有的话就是支持的
--
发件人：sunfulin
发送时间：2021年1月20日(星期三) 14:40
收件人：user-zh
主 题：Flink 1.11 SQL可以支持kafka动态分区发现么？
hi，
各位大神，请教下，1.11的sql作业，如何能实现动态分区发现呐？我在1.12的文档里发现有个参数可以设置，但是1.11的版本里貌似没有。想确认下能否支持？
--
发自我的网易邮箱手机智能版
hi，
各位大神，请教下，1.11的sql作业，如何能实现动态分区发现呐？我在1.12的文档里发现有个参数可以设置，但是1.11的版本里貌似没有。想确认下能否支持？
--
发自我的网易邮箱手机智能版
hi，
我遇到一个问题，消费的source里有字段定义为array>这种类型，然后想通过一个udf将它处理成一个字符串。udf的入参定义如下：
public String eval(Row[] item, String char1, String char2);
但是在函数处理时，debug发现拿到的item里的row信息始终为null。也通过DataTypeHint注解给出了item的实际类型。这是不是1.10的bug呀？如果有相关的issue单的话，烦请有知道的发我下哈。
我在1.11里验证同样的逻辑，是没这个问题的。
hi，
session window能处理这种超时事件么？不知道有没有例子可以参考参考哈。
在 2020-10-30 11:12:55，"naisili Yuan" 写道：
> 不知道理解错没有， 感觉你这个场景使用session windows能解决
>
>sunfulin 于2020年10月30日周五 上午11:01写道：
>
>> hi，community，
>> 我最近有一个业务场景，需要基于消息流和具体的业务逻辑判断生成超时事件，考虑通过flink
>> cep来
hi，community，
我最近有一个业务场景，需要基于消息流和具体的业务逻辑判断生成超时事件，考虑通过flink
cep来实现。不过在这个场景中，需要针对输入的消息，判断如果一个小时内没有匹配的数据到来，就需要把该事件输出。
目前的cep机制，应该需要下一个事件消息到来时才会输出事件。想请教下各位大神，针对这个诉求有没有啥好的方案。
感谢。
hi，
我执行如下的sql场景：select userId, first_value(xxx) from source group by userId,
date_format(eventtime, '-MM-dd')
在不使用minibatch时，出现状态丢失的情况，同一个userId同一天输出了多条记录。这种可能是bug么？使用的flink 版本为1.10.1
hi，community，
想确认下idlestateretention的配置及生效机制，我有一个作业，设置了TTL为（10小时，10小时+5分钟）。假设我的作业是今天12:00启动，作业逻辑是统计当日10点后每个userId第一次登陆的时间：select
userId, first_value(xxx) from source group by userId, date_format(eventtime,
'-MM-dd')。那么我的作业的状态清理时机是从启动时间开始10小时之后么？还是会按照状态的数据更新的时间+10小时作为清理时间？
我使用flink
hi, benchao,
感谢回复，那我是不是可以理解为：去掉minibatch，就可以状态过期清理了哈？
在 2020-08-14 14:09:33，"Benchao Li" 写道：
>Hi,
>现在group agg + mini batch 还没有支持状态过期清理，已经有工作[1] 在解决这个问题了。
>
>[1] https://issues.apache.org/jira/browse/FLINK-17096
>
>sunfulin 于2020年8月14日周五 下午2:06写道：
hi，我的一个flink sql作业，在启用了idlestateretentiontime设置后，观察到web ui上的state
size还是一直在增大，超过maximum retention time之后state大小也没有减小的情况，请问这个可能是啥原因哈？
使用的flink 版本：flink 1.10.1，启用的state
ttl配置：tableEnv.getConfig.setIdleStateRetentionTime(Time.minutes(5),
Time.minutes(10));
我的作业逻辑是：统计每个userId每天第一次出现的记录，类似：select
hi,
再问下，这个方案还是会提交两个job吧？
在 2020-07-17 14:36:19，"godfrey he" 写道：
>做不到，1.11里把 StreamExecutionEnvironment.execute 和
>StreamTableEnvironment.execute 的逻辑已经切分干净了。
>有个改动比较小的方案可以参考：可以在原来的逻辑的基础上，把两种提交job的方式放到两个不同的类中，其他的逻辑放到另外一个类共性。
>
>sunfulin 于2020年7月17日周五 下
，"sunfulin" 写道：
hi,
感谢回复。这个机制我理解了。想了解下，有办法在1.11里仍然使用1.10版本的作业提交机制么？我现在虽然把代码回滚到1.10版本的逻辑，但是提交作业仍然有问题：比如我如果不执行env.execute，那么table
to DataStream的语句不会生成拓扑。
在 2020-07-17 12:09:20，"godfrey he" 写道：
>hi sunfulin,
>目前这个做不到。executeSQL 和 table to DataStream 是分别优
hi,
感谢回复。这个机制我理解了。想了解下，有办法在1.11里仍然使用1.10版本的作业提交机制么？我现在虽然把代码回滚到1.10版本的逻辑，但是提交作业仍然有问题：比如我如果不执行env.execute，那么table
to DataStream的语句不会生成拓扑。
在 2020-07-17 12:09:20，"godfrey he" 写道：
>hi sunfulin,
>目前这个做不到。executeSQL 和 table to DataStream 是分别优化和提交作业的。
>即使在1.11 之前，table
hi，
请教下flink 1.11任务提交的问题。如果我的一个作业里既有sql
dml提交（executeSQL执行），又通过DataStream.addSink来写出，
通过StreamExecutionEnvironment.execute提交，yarn
per-job貌似会提交两个作业。这种情况下，我该如何处理呢？只想提交一个作业。
rmats/json.html#json-timestamp-format-standard
>
><https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/formats/json.html#json-timestamp-format-standard>
>
>> 在 2020年7月15日，23:19，sunfulin 写道：
>>
>> hi,
>> 我通过flink sql 定义了一个es sink，其中有个字
at
org.apache.flink.formats.json.JsonRowDataSerializationSchema.createNotNullConverter(JsonRowDataSerializationSchema.java:184)
在 2020-07-15 21:24:30，"sunfulin" 写道：
>hi，
>我看1.11的java.sql.Timestamp
>对应的是Flink的TIMESTAMP(9)，跟之前默认的TIMESTAMP(3)有区别，而且之前1.10的Timestamp(3)是带时区UTC的，
hi，
我看1.11的java.sql.Timestamp
对应的是Flink的TIMESTAMP(9)，跟之前默认的TIMESTAMP(3)有区别，而且之前1.10的Timestamp(3)是带时区UTC的，现在这个类型不带时区了。想问下这个具体调整应该如何适配？
hi,
@Danny Chan
我在1.10版本中确实触发到了这个bug，切到1.11版本貌似就没这问题了。简单解释下问题：双流join的case，右边流join后的结果字段在获取时貌似乱序了。
在 2020-07-13 10:42:12，"Jark Wu" 写道：
>cc @Danny Chan 也许 Danny 老师知道。
>
>On Thu, 9 Jul 2020 at 17:29, sunfulin wrote:
>
>>
>> hi,
>> 我切到最
quot;f1").as("firstx"), $("p").proctime());
//sink写入
StatementSet ss = tEnv.createStatementSet();
ss.addInsertSql("insert into es_sink_test_no_pk select idx, firstx from
test");
ss.addInsertSql("insert into es_sink_test_
hi，
感谢详细的解释和回复。那问题就清楚了。之前我们的job提交框架里统一都使用了StreamExecutionEnvironment.execute(jobName)方法，现在基于这个解释就明白了。
在 2020-07-12 22:55:34，"godfrey he" 写道：
>hi sunfulin，
>
>1.11 对 StreamTableEnvironment.execute()
>和 StreamExecutionEnvironment.execute()
sticsearch/blob/977230a0ce89a55515dc6ef6452e9f059d9356a2/core/src/main/java/org/elasticsearch/action/index/IndexRequest.java#L509
>>
>> Best,
>> Yangze Guo
>>
>> On Sat, Jul 11, 2020 at 11:33 PM sunfulin wrote:
>>>
>>> hi，
>>> 根据文档[1]的描述，1
uration
>> 相关的修改有关，具体到这个错误，你可以按照提示增加一些内存看看
>>
>> [1]
>>
>> https://flink.apache.org/news/2020/07/06/release-1.11.0.html#other-improvements
>> Best,
>> Congxian
>>
>>
>> sunfulin 于2020年7月10日周五 下午11:32写道：
>>
>> > hi，
>> >
&g
hi，
根据文档[1]的描述，1.11的es sql connector如果在ddl里没有声明primary
key，将会使用append模式sink数据，并使用es本身生成的id作为document_id。但是我在测试时发现，如果我的ddl里没有定义primary
key，写入时没有正确生成document_id，反而是将index作为id生成了。导致只有最新的一条记录。下面是我的ddl定义：
不确定是我配置使用的方式不对，还是确实存在bug。。
CREATE TABLE ES6_SENSORDATA_OUTPUT (
event varchar,
user_id
(ThreadPoolExecutor.java:624)
[?:1.8.0_201]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_201]
在 2020-07-11 22:24:51，"sunfulin" 写道：
>hi，
>我使用flink
>1.11提交sql作业，从JM日志中看到有如下异常。我的作业里会通过tEnv.executeSQL执行多个ddl语句，通过tEnv.createStatementSet
> add多个dml语句，并执行execute。
>如下异常可能原因是
hi，
我使用flink
1.11提交sql作业，从JM日志中看到有如下异常。我的作业里会通过tEnv.executeSQL执行多个ddl语句，通过tEnv.createStatementSet
add多个dml语句，并执行execute。
如下异常可能原因是啥呢？还有个问题，这个异常虽然抛出来了，但是作业还是正常启动执行了。这又是为何？是不是不推荐在作业里同时使用executeSQL和statementset.execute？
Caused by: org.apache.flink.util.FlinkRuntimeException: Cannot have more than
hi，
我在使用1.11版本在本地idea起一个作业时，并发为1，抛出了如下关于内存的异常。。问题是之前从来没有显示配置过taskmanager的memory参数，这是为何？
感觉由1.10升级到1.11问题还是挺多的。。我尝试增加了JVM参数，增加DirectMemory内存配置，还是没有作用，请教大神帮忙看下。
Exception in thread "main" java.lang.OutOfMemoryError: Could not allocate
enough memory segments for NetworkBufferPool (required (Mb):
样建立一个 issue 来追踪这个问题
>
>[1]
>https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/elasticsearch.html#connector-options
>Best,
>Congxian
>
>
>Leonard Xu 于2020年7月10日周五 上午11:33写道：
>
>> Hi，
>> 我理解作业应该失败才对，你本地可以复现吗？可以复现的话可以在社区建个issue。
&
hi，Leonard
是的。es集群服务不可用。我能观察到写入es失败，但是作业确实没失败。等到es集群服务恢复后，作业也正常了，但是故障期间的数据有丢失。
在 2020-07-10 11:16:17，"Leonard Xu" 写道：
>Hello, fulin
>
>> es服务挂掉时，这段时间写入es失败，但是没有积压，而是数据丢失了。这个应该不符合预期。想问下可能是啥原因造成的。
>
>es 服务挂掉是指es 集群不可用吗？那这时应该是写入es应该失败，作业也会失败，你说的没有积压是指什么呢？
>
>Best
>Leonard Xu
S Sink 是自己写的，还是用的社区的呢？社区的使用了哪个版本，以及配置是啥样的呢
>
>Best,
>Congxian
>
>
>sunfulin 于2020年7月10日周五 上午10:51写道：
>
>> hi，
>>
>> 我们现在使用flink消费kafka数据写到es，今天发现在默认设置下，es服务挂掉时，这段时间写入es失败，但是没有积压，而是数据丢失了。这个应该不符合预期。想问下可能是啥原因造成的。
hi，
我们现在使用flink消费kafka数据写到es，今天发现在默认设置下，es服务挂掉时，这段时间写入es失败，但是没有积压，而是数据丢失了。这个应该不符合预期。想问下可能是啥原因造成的。
hi,
我切到最新的1.11 release版本，跑同样的sql，没有抛出异常。想问下这有相关的issue么？想确认下原因。
在 2020-07-09 16:53:34，"sunfulin" 写道：
>hi,
>我使用flink 1.10.1
>blink-planner。运行以下SQL时，抛出异常。其中A和B分别是两个Kafka消息流。任务使用processtime。如果我把join的B表的select
> 具体字段名 修改为 select *，貌似就可以执行。但是拿到的B表字段顺序貌似是错乱的。请问
hi,
我使用flink 1.10.1
blink-planner。运行以下SQL时，抛出异常。其中A和B分别是两个Kafka消息流。任务使用processtime。如果我把join的B表的select
具体字段名 修改为 select *，貌似就可以执行。但是拿到的B表字段顺序貌似是错乱的。请问这个问题是bug么？
select A.recvTime, A.khh, A.live_id, A.fund_code as product_code, A.fund_name
as product_name, cast(B.balance as double) as balance
一些 factory 的时候出错了（找不到），可以看看对应的 module 的
>resources 文件下是否有对应的 resource 文件 Best, Congxian sunfulin sunfulin0...@163.com
>于2020年7月7日周二 下午6:29写道： hi, 我的pom文件本地执行时，scope的provided都是去掉的。 dependency
>groupIdorg.apache.flink/groupId
>artifactIdflink-table-planner-blink_${scala.binary.vers
2020-07-07 18:10:58，"Jark Wu" 写道：
>如果是在 IDEA 中运行的话，你看看 blink planner 这个依赖的 scope 是不是被 provided 掉了？ 去掉 provided
>再试试看？
>
>Best,
>Jark
>
>On Tue, 7 Jul 2020 at 18:01, sunfulin wrote:
>
>> hi,
>> @Jun Zhang 我一直使用的就是blink planner，这个jar包一直都有的。
>
;> hi.sunfulin
>> 你有没有导入blink的planner呢，加入这个试试
>>
>>
>> org.apache.flink
>>
>> flink-table-planner-blink_${scala.binary.version}
>> ${flink.version}
>>
>>
>>
>> sunfulin 于2020年7月7日周二 下午3:21写道：
>>
>>>
b.* from test a left join my_dim
FOR SYSTEM_TIME AS OF a.p AS b on a.first = b.userId");
//输出
tEnv.toAppendStream(table, Row.class).print("LookUpJoinJob");
env.execute("LookUpJoinJob");
在 2020-07-06 14:59:17，"Jark Wu" 写道：
Hi，
我使用目前最新的Flink 1.11 rc4来测试我的作业。报了如下异常：
org.apache.flink.table.api.TableExecution: Failed to execute sql
caused by : java.lang.IlleagalStateException: No ExecutorFactory found to
execute the application.
at
hi
抱歉忘记回复了。经过进一步调试发现，是因为定义的schema的column类型，与实际获取到的字段类型不一致导致。主要是在调试的过程中，ComplettedFuture.complete会吃掉这种类型不一致的异常，也不下发数据。看源码发现只会在timeout的时候才调用future.completeException。记录下。
在 2020-07-03 17:01:19，"forideal" 写道：
>Hi sunfulin：
>
> 我这么实现是可以的。
>public void
hi，
我在使用flink 1.10.1 blink
planner，通过扩展tablesourcesinkfactory和asynctablefunction扩展了一个维表，维表会初始化并调用rpc服务。
遇到一个比较诡异的问题，作业在执行如下join的sql时，没有任何输出，等一段时间后，抛出了异常：Caused by :
java.lang.Exception: Could not complete the stream element:
org.apache.flink.table.dataformat.BinaryRow caused by :
g()的结果可能不是我们想要的。
>
>你可以试下下面的query，query keys 对应es中的 id 就是
>commentId${keyDelimiter}commentContent, 这也应该是你需要的结果
>Select ROW(commentId, commentContent) from T
>group by commentId, commentContent
>
>祝好，
>Leonard Xu
>
>> 在 2020年6月28日，22:33，sunfulin 写道：
>>
，"Benchao Li" 写道：
>Hi，
>我好像没有看到哪里说不可以group by ROW类型呀，可以说下你使用的是哪个Flink版本，以及哪个planner么？
>能附上异常栈就更好啦。
>
>sunfulin 于2020年6月25日周四 下午4:35写道：
>
>> Hi，
>> 请教大家一个问题，我在一个sink table里定义了一个column A，类型是Row。
>> 在通过Flink SQL 做sink的时候，尝试group by A，报错说ROW类型无法group by。这种应该如何处理哈？
>
>
>
>--
>
>Best,
>Benchao Li
Hi，
请教大家一个问题，我在一个sink table里定义了一个column A，类型是Row。
在通过Flink SQL 做sink的时候，尝试group by A，报错说ROW类型无法group by。这种应该如何处理哈？
识到了。
>目前已经有一个 FLIP [1] 在讨论中，预计 1.12 会支持。
>
>Best,
>Jark
>
>[1]:
>https://cwiki.apache.org/confluence/display/FLINK/FLIP-107%3A+Reading+table+columns+from+different+parts+of+source+records
>
>On Fri, 5 Jun 2020 at 19:19, sunfulin wrote:
>
>> Hi，
>> 想问下Flink S
Hi，
想问下Flink SQL在使用DDL创建Kafka
Source时，支持设置获取到Kafka自带的timestamp么？我们有场景想使用Kafka带的timestamp,这种情况下消息流中可能并不存在时间属性.
如果支持的话,能否分享下具体写法哈?我尝试使用下面的SQL报错：
CREATE TABLE user_behavior (
test_time TIMESTAMP(3),
user_id STRING ,
item_id STRING ,
category_id STRING ,
behavior STRING,
ts STRING,
proctime as
Hi,
props.put("auto.offset.reset", "latest");
是加了这个设置导致的吧
在 2020-04-07 11:27:53，"苟刚" 写道：
>Hello,
>
> 我遇到一个问题，我用flink做实时统计的时候，每次新增一种计算类型，算子就会从kafka的最早的消息开始消费，导致我每次重启后都需要花费好长的时间去追平记录，请问有什么办法解决吗？
>我的wartermark是设置在kafka的consumer上的，下面的每新增一个process的时候都会从头开始消费。
>
>
hi,
请教下，现有功能里，可以将hive表作为维表join么？作为temporal table。
如果可以的话，hive分区表如何join呢？一般维表join是要join最新分区的全量数据。
在 2020-04-02 17:30:39，"111" 写道：
>Hi,
>只要能解决upsert问题带来的存储碎片、读写冲突、版本回溯，实时写入Hive也是可以的，目前spark delta lake就已经做到了。
>前面jingsong也提到过，会去解决文件存储、合并等问题，那到时候flink实时写入hive就没问题了。
>Best,
>Xinghalo
Hi，
这里就涉及到一个话题，应该怎么去实践实时和离线数仓的数据融合。我能理解这个技术上的不合理，但是通过Flink实现数仓ETL的功能，我相信很多使用Flink的会将之作为一个重要场景。
在 2020-04-01 16:05:54，"111" 写道：
>
>
>Hi,
>流写入hive，其实是属于数据湖的概念范畴。
>因为流往hive里面写，会造成很多的碎片文件，对hdfs造成性能影响，因此一般不会在流场景下直接写入hive。
>详细的可以了解 Delta lake 或 hudi。
>
&
持这类场景。
你可以描述下详细堆栈、应用场景、SQL吗？
Best,
Jingsong Lee
On Wed, Apr 1, 2020 at 2:56 PM sunfulin wrote:
我使用batch mode时，又抛出了如下异常：感觉一步一个坑。。sigh
org.apache.calcite.plan.RelOptPlanner$CannotPlanException: There are not enough
rules to produce a node with desired properties
在 2020-0
式中。功能正在开发中[1]
>
>[1]
>https://cwiki.apache.org/confluence/display/FLINK/FLIP-115%3A+Filesystem+connector+in+Table
>
>Best,
>Jingsong Lee
>
>On Wed, Apr 1, 2020 at 2:32 PM sunfulin wrote:
>
>> Hi,
>> 我这边在使用Flink消费Kafka数据写入hive。配置连接都OK，但是在实际执行insert into
>&
Hi,
我这边在使用Flink消费Kafka数据写入hive。配置连接都OK，但是在实际执行insert into
xxx_table时，报了如下异常。这个看不懂啥原因，求大神指教。
cc @Jingsong Li @Jark Wu
org.apache.flink.table.api.TableException: Stream Tables can only be emitted by
AppendStreamTableSink, RetractStreamTableSink, or UpsertStreamTableSink.
at
Hi，
据我所知现在还不能直接支持Oracle的driver吧？你是怎么使用Flink SQL读写oracle的哈？
在 2020-03-27 17:21:21，"111" 写道：
>Hi,
>在使用Flink SQL读写Oracle JDBC表时，遇到了timestamp转换异常：
>Caused by: java.lang.ClassCastException: oracle.sql.TIMESTAMP cannot be cast
>to java.sql.Timestamp at
这样来用：
StreamTableEnvironment.getConfig().setIdleStateRetentionTime(min, max);
在 2020-03-12 14:11:31，"wangl...@geekplus.com.cn" 写道：
>
>这个文档是最新的吗，我直接在 IDEA 里面写这三行代码。
>StreamQueryConfig Deprecated, tableEnv 没有 queryConfig() 方法
>StreamExecutionEnvironment env =
rror message, it seems that this issue[1] is similar with
>yours, but it seems that current compile util does not have this issue.
>
>BTW, do you using 1.10?
>
>[1] https://issues.apache.org/jira/browse/FLINK-7490
>
>sunfulin 于2020年3月2日周一 上午11:17写道：
>
>>
>>
>&g
ot; wrote:
>Could you also provide us the DDL for lscsp_sc_order_all
>and dim_app_cust_info ?
>
>sunfulin 于2020年3月1日周日 下午9:22写道：
>
>>
>> *CREATE TABLE **realtime_product_sell *(
>> sor_pty_id *varchar*,
>> entrust_date *varchar*,
>> entrust_time *var
DF looks good. Could you also paste your DDL? Then we can produce your
>bug easily.
>
>sunfulin 于2020年3月1日周日 下午6:39写道：
>
>> Below is the code. The function trans origin field timeStr "2020-03-01
>> 12:01:00.234" to target timeStr accroding to dayTag.
>&g
t(format);
return sf.format(date);
}
}
}
At 2020-03-01 18:14:30, "Benchao Li" wrote:
Could you show how your UDF `ts2Date` is implemented?
sunfulin 于2020年3月1日周日 下午6:05写道：
Hi, Benchao,
Thanks for the reply.
Could you provide us more information?
1. what planner are you using? blink or
ointing? what if you enable
checkpointing and not use your udf? and disable checkpointing and use udf?
sunfulin 于2020年3月1日周日 下午5:41写道：
Hi, guys
I am running Flink 1.10 SQL job with UpsertStreamSink to ElasticSearch. In my
sql logic, I am using a UDF like ts2Date to handle date format stream fiel
Hi, guys
I am running Flink 1.10 SQL job with UpsertStreamSink to ElasticSearch. In my
sql logic, I am using a UDF like ts2Date to handle date format stream fields.
However, when I add the `env.enableCheckpointing(time)`, my job failed to
submit and throws exception like following. This is
: flink_t...@hadoop.htsc.com
At 2020-02-21 18:18:57, "sunfulin" wrote:
Hi，
我使用Flink
1.10集成hive，在连接metastore的时候由于hive对应CDH集群开启了kerberos认证，抛出了如下异常：请问大家这个该怎么配置或者解决哈？
999 [main] INFO hive.metastore - Trying to connect to metastore with URI
thrift://namenode01.htsc.com:9083
Hi，
我使用Flink
1.10集成hive，在连接metastore的时候由于hive对应CDH集群开启了kerberos认证，抛出了如下异常：请问大家这个该怎么配置或者解决哈？
999 [main] INFO hive.metastore - Trying to connect to metastore with URI
thrift://namenode01.htsc.com:9083
1175 [main] ERROR org.apache.thrift.transport.TSaslTransport - SASL
negotiation failure
好的，感谢。我在user里附加了query SQL。
在 2020-02-15 16:14:56，"Jark Wu" 写道：
>Hi sunfulin,
>
>这个异常是说通过 query 推断不出 query 的 primary key，不是说 sink 没有 primary key。至于为什么 query
>推断不出 pk，可能要结合 query 看一下。
>我看到你在 user@ 里面也发邮件了，我已经在那下面回复了，我们要不在 user@ 邮件下面继续讨论吧。可以将你们的 SQL 补充一下，包括
>D
grading? There were some changes in
>1.9 for how to parse timestamps in JSON format.
>
>Your error might be related to those changes:
>
>https://issues.apache.org/jira/browse/FLINK-11727
>
>I hope this helps.
>
>Timo
>
>
>On 07.02.20 07:57, sunfulin wrote
Hi ,guys
Thanks for kind reply. Actually I want to know how to change client side haddop
conf while using table API within my program. Hope some useful sug.
At 2020-02-11 02:42:31, "Bowen Li" wrote:
Hi sunfulin,
Sounds like you didn't config the hadoop HA correctly on
Hi, guys
I am using Flink 1.10 and test functional cases with hive intergration. Hive
with 1.1.0-cdh5.3.0 and with hadoop HA enabled.Running flink job I can see
successful connection with hive metastore, but cannot read table data with
exception:
java.lang.IllegalArgumentException:
Hi, guys
When upgrading to Flink 1.10 rc0, I am confused by the new DataTypes schema
defination.
I am reading and consuming records from kafka with json schema like
{"user_id":1866071598998,"recv_time":1547011832729}. and the code snippet is :
.withSchema(
new Schema()
//
new Json().failOnMissingField(false)
.deriveSchema()
)
.inAppendMode()
.registerTableSource(getTableName());
在 2020-01-10 09:53:52，"Evan" 写道：
>第一，查看神策的kafka的配置项advertised.host.name
>
>
>
>
>--原始邮件--
>发件人:&
我有一个作业，通过Flink去消费神策的Kafka消息数据，参见（https://manual.sensorsdata.cn/sa/latest/page-1573828.html）。但是任务启动之后，Flink任务没有消费任何Kafka消息，我通过神策的kafka
console consumer能看到源源不断的数据。
本身flink作业job里没有任何报错，想问下大家这种情况该如何排查？
wrote:
>hi sunfulin,
>
>As Kurt pointed out, if you use RocksDB state backend, maybe slow disk IO
>bound your job.
>You can check WindowOperator's latency metric to see how long it tasks to
>process an element.
>Hope this helps.
>
>sunfulin 于2020年1月8日周三 下午4:04写道：
>
&g
求问怎么通过dashboard查看状态存储量之类的统计？如果没开checkpoint的话
:53 PM 贺小令 wrote:
hi sunfulin,
you can try with blink planner (since 1.9 +), which optimizes distinct
aggregation. you can also try to enable
table.optimizer.distinct-agg.split.enabled if the data is skew.
best,
godfreyhe
sunfulin 于2020年1月8日周三 下午3:39写道：
Hi, community,
I'm using Apache
.
For now I am not able to use blink planner on my apps because the current prod
environment has not planned or ready to up to Flink 1.9+.
At 2020-01-08 15:52:28, "贺小令" wrote:
hi sunfulin,
you can try with blink planner (since 1.9 +), which optimizes distinct
aggregation. you ca
Hi, community,
I'm using Apache Flink SQL to build some of my realtime streaming apps. With
one scenario I'm trying to count(distinct deviceID) over about 100GB data set
in realtime, and aggregate results with sink to ElasticSearch index. I met a
severe performance issue when running my flink
71 matches
Mail list logo
