flink sql作业从kafka消费mysql过来的canal
json消息，经过复杂处理后写入doris，请问如何统计doris表记录的端到端时延？mysql表有update_time字段代表业务更新记录时间。
doris系统可以在表schema新增一个更新时间列ingest_time，所以在doris表上可以通过ingest_time -
update_time算出端到端时延，但这种方法只能离线统计，有没有实时统计以方便实时监控的方法？
感谢！那是不是要计算sink延迟的话只需用 context.currentProcessingTime() - context.timestamp() 即可？
我看新的sink
v2中的SinkWriter.Context已经没有currentProcessingTime()方法了，是不是可以直接使用System.currentTimeMillis()
- context.timestamp()得到sink延迟呢？
在 2024-02-21 09:41:37，"Xuyang" 写道：
>Hi, chen.
>可以试一下在sink
Hi Lorenzo,
I think the most convenient way is to modify the code of the state backend,
adding a k-v cache as you want.
Otherwise IIUC, there's no public interface to get keyContext. But well,
you may try something hacky. You may use the passed-in `Context` instance
in processElement, and
Hi all,
The Apache Kyuubi community is pleased to announce that
Apache Kyuubi 1.8.1 has been released!
Apache Kyuubi is a distributed and multi-tenant gateway to provide
serverless SQL on data warehouses and lakehouses.
Kyuubi provides a pure SQL gateway through Thrift JDBC/ODBC and
RESTful
Hello Guys,
Can someone please assist us regarding the following issue ?
We have noticed that when we add a *new kafka sink* operator to the graph, *and
start from the last save point*, the operator is 100% busy for several
minutes and *even 1/2-1 hour* !!!
The problematic code seems to be the
Hi, chen.
可以试一下在sink function的invoke函数中使用：
@Override
public void invoke(RowData row, Context context) throws Exception {
context.currentProcessingTime();
context.currentWatermark();
...
}
--
Best！
Xuyang
在 2024-02-20 19:38:44，"Feng Jin"
Hi Arjun,
Yes, direct support for external configuration files within Flink
ConfigMaps is somewhat restricted. The current method involves simply
copying two local files from the operator.
Please check : FlinkConfMountDecorator#getLocalLogConfFiles()
Thanks Zakelly,
I'd need to do something similar, with a map containing my non-serializable
"state", similar to the kvCache in FastTop1Fucntion.
But I am not sure I understand how I can set the keyed state for a specific
key, in snapshotState().
FastTop1Function seems to rely on keyContext set
Classification: External
Hi,
I have a use case involving calculating the lifetime order count of a
customer in real-time. To reduce the memory footprint, I plan to run a
batch job on stored data every morning (let's say at 5 am) to calculate the
total order count up to that moment. Additionally,
我理解不应该通过 rowData 获取， 可以通过 Context 获得 watermark 和 eventTime.
Best,
Feng
On Tue, Feb 20, 2024 at 4:35 PM casel.chen wrote:
> 请问flink sql中的自定义sink connector如何获取到source table中定义的event time和watermark?
>
>
> public class XxxSinkFunction extends RichSinkFunction implements
> CheckpointedFunction,
Hello to all users, contributors and Committers!
The Travel Assistance Committee (TAC) are pleased to announce that
travel assistance applications for Community over Code Asia 2024 are now
open!
We will be supporting Community over Code Asia, Hangzhou, China
July 26th - 28th, 2024.
TAC exists
退订
请问flink sql中的自定义sink connector如何获取到source table中定义的event time和watermark?
public class XxxSinkFunction extends RichSinkFunction implements
CheckpointedFunction, CheckpointListener {
@Override
public synchronized void invoke(RowData rowData, Context context) throws
IOException {
13 matches
Mail list logo
