Hi
请教一个问题
我需要对一个类似这样的数据进行计算获取用户 categoryId
| userId | articleID | categoryId | score |
| 01 | A | 1 | 10 |
| 01 | B | 1 | 20 |
| 01 | C | 2 | 30 |
目前我的实现是使用tableAPI 根据 UserId和categoryID 分组做 score 聚合 再通过状态做TopN排序,有没有其他更好的方案来实现?
我使用过 ROW_NUMBER() over() ,在 flink 1.10 上,并无法使用。分组聚合除了使用Table AP
DataStream??apiUV??2
1Tumbling??1Time.days(1)??uv
trigger
2???
Hi xuhaiLong,
看下是否能够在 Flink SQL 中,通过自定义 UDTF 来满足你的需求。
Best,
LakeShen
xuhaiLong 于2020年6月23日周二 下午7:18写道:
> Hi
>
> 请教一个问题
>
>
> 我需要对一个类似这样的数据进行计算获取用户 categoryId
> | userId | articleID | categoryId | score |
> | 01 | A | 1 | 10 |
> | 01 | B | 1 | 20 |
> | 01 | C | 2 | 30 |
>
>
>
>
> 目前我的实现是使用table
Hi xuhaiLong,
1.10 blink planner 是支持 ROW_NUMBER() over 的 (配合 rownum <= N 使用)。你是不是用的 old
planner 呢?
Best,
Jark
On Tue, 23 Jun 2020 at 19:44, LakeShen wrote:
> Hi xuhaiLong,
>
> 看下是否能够在 Flink SQL 中,通过自定义 UDTF 来满足你的需求。
>
> Best,
> LakeShen
>
> xuhaiLong 于2020年6月23日周二 下午7:18写道:
>
> > Hi
> >
> > 请
使用的是1.10.1,在 table api 无法使用ROW_NUMBER
On 6/23/2020 19:52,Jark Wu wrote:
Hi xuhaiLong,
1.10 blink planner 是支持 ROW_NUMBER() over 的 (配合 rownum <= N 使用)。你是不是用的 old
planner 呢?
Best,
Jark
On Tue, 23 Jun 2020 at 19:44, LakeShen wrote:
Hi xuhaiLong,
看下是否能够在 Flink SQL 中,通过自定义 UDTF 来满足你的需求。
Best,
Lake
图片无法查看,你可以把图片上传到某图床,然后将链接贴这里。
On Tue, 23 Jun 2020 at 19:59, xuhaiLong wrote:
> 使用的是1.10.1,在 table api 无法使用ROW_NUMBER
> On 6/23/2020 19:52,Jark Wu wrote:
>
> Hi xuhaiLong,
>
> 1.10 blink planner 是支持 ROW_NUMBER() over 的 (配合 rownum <= N 使用)。你是不是用的 old
> planner 呢?
>
> Best,
> Jark
>
> On Tue, 23 J
"org.apache.flink" %% "flink-table-api-scala-bridge" % "1.10.1",
"org.apache.flink" %% "flink-table-planner" % "1.10.1" % "provided",
看下粘贴的 sbt 依赖
On 6/23/2020 20:06,Jark Wu wrote:
图片无法查看,你可以把图片上传到某图床,然后将链接贴这里。
On Tue, 23 Jun 2020 at 19:59, xuhaiLong wrote:
使用的是1.10.1,在 table api 无法使用ROW_NUMBER
Hi xuhaiLong,
看你的依赖,应该用的 old planner,你要使用 blink planner 才能使用row_number 函数。要使用
flink-table-planner-blink_2.11
具体文档参考:
https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/#table-program-dependencies
Best,
LakeShen
xuhaiLong 于2020年6月23日周二 下午8:14写道:
> "org.apache.flink" %% "flin
是我的问题,引用了old planner。感谢!
On 6/23/2020 21:05,LakeShen wrote:
Hi xuhaiLong,
看你的依赖,应该用的 old planner,你要使用 blink planner 才能使用row_number 函数。要使用
flink-table-planner-blink_2.11
具体文档参考:
https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/#table-program-dependencies
Best,
LakeShen
xuh
想在Flink DataStream中实现微批,可以做到吗。该如何实现?
比方说基于处理时间,从Kafka 消费数据,每隔5s产生一个Batch,然后对这个Batch处理。
感谢!
请教一下, Flink DataStream中有广播变量的概念吗。
比方说,我有一个全局变量ArrayList,在JobManager端更新,然后在TaskManager中共享。可以做到吗
感谢!
用 Window 实现。
2019年10月22日 +0800 13:25 王佩 ,写道:
> 想在Flink DataStream中实现微批,可以做到吗。该如何实现?
>
> 比方说基于处理时间,从Kafka 消费数据,每隔5s产生一个Batch,然后对这个Batch处理。
>
> 感谢!
你好,
https://www.jianshu.com/p/c8c99f613f10?from=singlemessage&isappinstalled=0
希望可以帮助你
> 在 2019年10月22日,下午2:04,王佩 写道:
>
> 请教一下, Flink DataStream中有广播变量的概念吗。
>
> 比方说,我有一个全局变量ArrayList,在JobManager端更新,然后在TaskManager中共享。可以做到吗
>
> 感谢!
Flink 写入 Elasticsearch,Checkpoint 一直处于IN_PROGRESS 状态,从而导致数据不能写入ES。
如图:
[image: image.png]
帮忙看下!
感谢!
请教下:
1、DataStream 如按用户ID KeyBy后,同一个用户ID的数据最终会被分到一个Partition中吗?
2、假设1成立,这样就会有数据倾斜的问题。该如何解决?
3、假设1成立,如: DataStream
.keyBy(userID)
.timeWindow()
.aggregate(new
AggregateFunction(...)),这里的AggregateFu
你的图挂了,如果 checkpoint 有问题的话,可以先参考文章[1] 自查一下,如果还有问题,可以在邮件列表更新
[1] https://mp.weixin.qq.com/s/0jTVXa9ktyLynwvRelRYYg
Best,
Congxian
王佩 于2019年10月23日周三 上午8:38写道:
> Flink 写入 Elasticsearch,Checkpoint 一直处于IN_PROGRESS 状态,从而导致数据不能写入ES。
>
> 如图:
>
> [image: image.png]
>
> 帮忙看下!
> 感谢!
>
mber 9, 2019 11:45 AM
> 收件人: user-zh
> 主题: Flink DataStream KeyedStream 与 AggregateFunction
>
> 请教下:
>
> 1、DataStream 如按用户ID KeyBy后,同一个用户ID的数据最终会被分到一个Partition中吗?
>
> 2、假设1成立,这样就会有数据倾斜的问题。该如何解决?
>
> 3、假设1成立,如: DataStream
>.keyB
是否可以本地先聚合?
>> 3,AggFunc是指定做何种聚合,是sum, 还是avg, 还是count。不指定的话,Flink哪里指导你要计算啥?
>>
>> -邮件原件-
>> 发件人: 王佩
>> 发送时间: Saturday, November 9, 2019 11:45 AM
>> 收件人: user-zh
>> 主题: Flink DataStream KeyedStream 与 AggregateFunction
>>
>> 请教下:
>>
oujun 于2019年11月9日周六 下午7:46写道:
>
>> 1, 是
>> 2,没有标准答案,是否可以本地先聚合?
>> 3,AggFunc是指定做何种聚合,是sum, 还是avg, 还是count。不指定的话,Flink哪里指导你要计算啥?
>>
>> -邮件原件-
>> 发件人: 王佩
>> 发送时间: Saturday, November 9, 2019 11:45 AM
>> 收件人: user-zh
>> 主题: Flink Data
1, 是
2,没有标准答案,是否可以本地先聚合?
3,AggFunc是指定做何种聚合,是sum, 还是avg, 还是count。不指定的话,Flink哪里指导你要计算啥?
-邮件原件-
发件人: 王佩
发送时间: Saturday, November 9, 2019 11:45 AM
收件人: user-zh
主题: Flink DataStream KeyedStream 与 AggregateFunction
请教下:
1、DataStream 如按用户ID KeyBy后,同一个用户ID的数据最终会被分到一个Partition中吗?
2、假设1成立,这样就
Hi Jiazhi,
1.如果数据流量不是很大的话,按每条数据触发也没问题。另外,基于事件时间的情况,提前触发可以选择ContinuousEventTimeTrigger,可以查看Trigger接口的实现找到你想要的trigger。
2.窗口结束后会自动释放。一般对于Global窗口需要手动设置TTL
Best,
shizk233
ゞ野蠻遊戲χ 于2020年7月7日周二 下午10:27写道:
> 大家好!
>
> 想问下,现在在用DataStream的api来统计每天的UV,代码如下,有2个使用问题:
> 1、在使用Tumbling窗口的时候,由于使用窗口跨度是1天(Ti
你这个需求貌似是要看一天的 UV 的实时更新量,可以看一下 sliding window。如果是每天 0 点清零,实时看今天的
UV,那就是另一个问题了,应该需要自己定义 trigger & evictor
每条触发一次 window...看你数据量吧
Best,
tison.
shizk233 于2020年7月10日周五 上午10:23写道:
> Hi Jiazhi,
>
>
> 1.如果数据流量不是很大的话,按每条数据触发也没问题。另外,基于事件时间的情况,提前触发可以选择ContinuousEventTimeTrigger,可以查看Trigger接口的实现找到你想要的tri
各位大佬好:
想请教下,在使用mysql-cdc到iceberg,通过sql方式可以实现自动更新和删除功能。但在使用datastream
api进行处理后,注册成临时表,怎么实现类似于sql方式的自动更新和删除呢?
Hi all,
我在使用scala 开发streaming应用时,使用了AggregateFunction with a
ProcessWindowFunction,flink 版本是1.11.1,但是idea报错:
Error:(80, 8) overloaded method value aggregate with alternatives:
很奇怪为什么不能重载下面的方法
def aggregate[ACC: TypeInformation, V: TypeInformation, R:
TypeInformation](preAggregator: Aggregate
Hi all,
我在使用scala 开发streaming应用时,使用了AggregateFunction with a
ProcessWindowFunction,flink 版本是1.11.1,但是idea报错:
Error:(80, 8) overloaded method value aggregate with alternatives:
很奇怪为什么不能重载下面的方法
def aggregate[ACC: TypeInformation, V: TypeInformation, R:
TypeInformation](preAggregator: Aggregate
在Flink 写ES,当ES集群繁忙时,会有如下异常:
2019-09-17 16:01:02 ERROR
org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase
afterBulk 430 Failed Elasticsearch bulk request: Connection closed
org.apache.http.ConnectionClosedException: Connection closed
at
org.apache.http.nio.protocol.HttpAsyncR
哈喽,请问你这个问题解决了吗?我也遇到了同样的问题...
--
Sent from: http://apache-flink.147419.n8.nabble.com/
on/
发件人: zapjone
日期: 星期二, 2024年6月4日 18:34
收件人: user-zh@flink.apache.org
主题: Flink Datastream实现删除操作
各位大佬好:
想请教下,在使用mysql-cdc到iceberg,通过sql方式可以实现自动更新和删除功能。但在使用datastream
api进行处理后,注册成临时表,怎么实现类似于sql方式的自动更新和删除呢?
有个疑问,
如下程序片段:
--
Properties properties = new Properties();
properties.setProperty("bootstrap.servers",kafkaAddr);
properties.setProperty("group.id",kafkaOdsGroup);
properties.setProperty("auto.offset.reset",kafkaOdsAutoOffsetReset);
properties.setProperty(Flin
个人理解是不会重复驱动执行的,具体你可以测试一下,但是从底层原理上讲,我也讲不了。
发件人: lp
发送时间: 2021-03-05 17:31
收件人: user-zh
主题: flink dataStream多次sink DAG重复驱动执行?
有个疑问,
如下程序片段:
--
Properties properties = new Properties();
properties.setProperty("bootstrap.servers",kafkaAddr);
properties.setProperty
一个Flink DataStream 作业从mysql cdc消费处理后写入apache
doris,请问有没有办法(从JobGraph/StreamGraph)获取到source/sink connector信息,包括连接字符串、数据库名、表名等?
Hi ,请教一个奇怪的问题:
streamSource.flatMap(new ComeIntoMaxFlatMapFunction())
.assignTimestampsAndWatermarks(new
CommonAssignerPeriodWatermarks<>(Time.seconds(1).toMilliseconds()))
.connect(ruleConfigSource)
.process(new MetricDataFilterProcessFunction())
.keyBy((KeySelector) metric -> {
请教一下,在idea中用flink datastream
api写的代码,source输入是一条一条socket流数据,那如何在本地idea中进行调试,观察每条输入数据的运行情况,idea是否支持这种调试?
,
Feng
On Mon, Feb 26, 2024 at 6:20 PM casel.chen wrote:
> 一个Flink DataStream 作业从mysql cdc消费处理后写入apache
> doris,请问有没有办法(从JobGraph/StreamGraph)获取到source/sink
> connector信息,包括连接字符串、数据库名、表名等?
应该是 window 还没达到触发的条件,可以看下 watermark 是否在推进
在 2020-11-18 15:29:54,"huang botao" 写道:
>Hi ,请教一个奇怪的问题:
>
>streamSource.flatMap(new ComeIntoMaxFlatMapFunction())
>
>.assignTimestampsAndWatermarks(new
>CommonAssignerPeriodWatermarks<>(Time.seconds(1).toMilliseconds()))
>
>.connect(ruleConfigSour
感谢您的回复,是这样的,我这边的环境设置用的是eventTime
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
window设置的是 slid(2,1), 但是等了足够长的时间还是不能触发 apply 方法
On Wed, Nov 18, 2020 at 5:50 PM hailongwang <18868816...@163.com
可以检查一下作业消费的 kafka 分区是否都有数据,如果有的分区无数据的话,那么可能会导致水印不会更新,从而窗口触发不了。
Best
zhisheng
huang botao 于2020年11月18日周三 下午10:34写道:
> 感谢您的回复,是这样的,我这边的环境设置用的是eventTime
>
> StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> env.setStreamTimeCharacteristic(TimeCharacteristic
hi, zhisheng, hailongwang:
感谢对这个问题的解答,这个问题确实出在了window无法触发的地方,原因是 在connect()
后面没有定义watermar导致,在connect后指定watermark就可以触发window了。
On Wed, Nov 18, 2020 at 10:46 PM zhisheng wrote:
> 可以检查一下作业消费的 kafka 分区是否都有数据,如果有的分区无数据的话,那么可能会导致水印不会更新,从而窗口触发不了。
>
> Best
> zhisheng
>
> huang botao 于2020年11月18日周三
connect前生成watermark也是可以的应该,但是你需要把ruleConfigSource流也赋watermark。我猜是这个地方出问题了。
huang botao 于2020年11月19日周四 下午12:58写道:
> hi, zhisheng, hailongwang:
>
> 感谢对这个问题的解答,这个问题确实出在了window无法触发的地方,原因是 在connect()
> 后面没有定义watermar导致,在connect后指定watermark就可以触发window了。
>
>
>
> On Wed, Nov 18, 2020 at 10:46 PM zhishen
支持的, 在 idea 中执行 main 函数即可.执行前,idea 中的运行配置中,最好勾选上: *Include dependencies
with "Provided" scope *否则有可能会有 class not found 的报错.
Best,
Feng Jin
On Sat, Apr 22, 2023 at 9:28 PM m18751805115_1 <18751805...@163.com> wrote:
> 请教一下,在idea中用flink datastream
> api写的代码
g Jin |
> | 日期 | 2023年04月22日 21:53 |
> | 收件人 | user-zh@flink.apache.org |
> | 抄送至 | |
> | 主题 | Re: flink datastream api写的代码如何在idea中调试 |
> 支持的, 在 idea 中执行 main 函数即可.执行前,idea 中的运行配置中,最好勾选上: *Include dependencies
> with "Provided" scope *否则有可能会有 class not found 的报错.
>
/OpenLineage/OpenLineage/blob/main/integration/flink/shared/src/main/java/io/openlineage/flink/visitor/wrapper/FlinkKafkaConsumerWrapper.java
Best,
Feng
On Mon, Feb 26, 2024 at 6:20 PM casel.chen wrote:
一个Flink DataStream 作业从mysql cdc消费处理后写入apache
doris,请问有没有办法(从JobGraph/StreamGraph)获取到source/sink
日 16:18,Zhanghao Chen 写道:
JobGraph 里有个字段就是 jobid。
Best,
Zhanghao Chen
From: 阿华田
Sent: Friday, March 8, 2024 14:14
To: user-zh@flink.apache.org
Subject: 回复: Flink DataStream 作业如何获取到作业血缘?
获取到Source 或者 DorisSink信息之后, 如何知道来自那个flink任务,好像不能获取到flinkJobId
| |
阿华田
From: 阿华田
Sent: Friday, March 8, 2024 16:48
To: user-zh@flink.apache.org
Subject: 回复: Flink DataStream 作业如何获取到作业血缘?
”JobGraph 可以获得 transformation 信息“, JobGraph可以直接获取transformation的信息吗?, 我们是在
SourceTransformation 和SinkTransformation反射拿到链接信息 ,但是这个地方拿不到flinkJobid,
JobGraph可以拿到source和
抱歉啊,可能我没有把问题描述清楚。我是想本地对代码进行debug调试,观察每条流输入后的变量值以及调用栈等信息的。
回复的原邮件
| 发件人 | Feng Jin |
| 日期 | 2023年04月22日 21:53 |
| 收件人 | user-zh@flink.apache.org |
| 抄送至 | |
| 主题 | Re: flink datastream api写的代码如何在idea中调试 |
支持的, 在 idea 中执行 main 函数即可.执行前,idea 中的运行配置中,最好勾选上: *Include dependencies
with
thx
回复的原邮件
| 发件人 | Feng Jin |
| 日期 | 2023年04月22日 22:22 |
| 收件人 | user-zh@flink.apache.org |
| 抄送至 | |
| 主题 | Re: flink datastream api写的代码如何在idea中调试 |
如果你是要本地 idea debug 线上的作业,需要在 taskManager 的 JVM 参数中开启debug
提交作业时, 添加参数:
env.java.opts.taskmanager="-agentlib:jdwp=transport=dt_s
JobGraph 里有个字段就是 jobid。
Best,
Zhanghao Chen
From: 阿华田
Sent: Friday, March 8, 2024 14:14
To: user-zh@flink.apache.org
Subject: 回复: Flink DataStream 作业如何获取到作业血缘?
获取到Source 或者 DorisSink信息之后, 如何知道来自那个flink任务,好像不能获取到flinkJobId
| |
阿华田
|
|
a15733178...@163.com
|
签名
/integration/flink/app/src/main/java/io/openlineage/flink/OpenLineageFlinkJobListener.java
Best,
Zhanghao Chen
From: 阿华田
Sent: Friday, March 8, 2024 16:48
To: user-zh@flink.apache.org
Subject: 回复: Flink DataStream 作业如何获取到作业血缘?
”JobGraph 可以获得 transformation 信息
From: 阿华田
Sent: Friday, March 8, 2024 18:47
To: user-zh@flink.apache.org
Subject: 回复: Flink DataStream 作业如何获取到作业血缘?
我们想修改源码 实现任意任务提交实时平台,初始化DAG的时候获取到血缘信息,StreamExecutionEnvironment注册 这种只能写在任务里
不满足需求
| |
阿华田
|
|
a15733178...@163.com
|
签名由网易邮箱大师定制
在2024年03月8日 18:23
49 matches
Mail list logo