How can I set job parameter in flink sql

Hi dear engineer, I want to override the function open() in my UDF, like: | public class BlackListConvertFunction extends ScalarFunction { @Override public void open(FunctionContext context) throws Exception { String path = context.getJobParameter("black_list_path", "/confi

How can I set job parameter in flink sql

Hi dear engineer, I want to override the function open() in my UDF, like: In open() function, I want to fetch the configred value "black_list_path", then simply print that value out. And I config this value in ./sql-client.sh console: SET black_list_path = /root/list.properties Then

Getting JSON kafka source parsing error with DDL

Hi Everyone, I am trying to fix the flink-playground for version 1.14.4 and was working on fixing pyflink-walkthrough and I getting following error Caused by: org.apache.flink.table.api.NoMatchingTableFactoryException: Could not find a suitable table factory for 'org.apache.flink.table.factories.

Re: Converting from table to stream, following Avro schema

Hi Dhavan, The type of the `ds` appearing in `t_env.from_data_stream(ds) should be known. Otherwise, it's impossible to infer the schema of the converted table, as a result, `raw` type will be used which makes the schema of the resulting table not expected. You could either declare the type of th

[ANNOUNCE] Apache Flink Table Store 0.1.0 released

The Apache Flink community is very happy to announce the release of Apache Flink Table Store 0.1.0. Apache Flink Table Store provides storage for building dynamic tables for both stream and batch processing in Flink, supporting high speed data ingestion and timely data query. Please check out the

Re: [Discuss] Creating an Apache Flink slack workspace

I agree with Robert on reworking the "Community" and "Getting Help" pages to emphasize how we position the mailing lists and Slack, and on revisiting in 6-12 months. Concerning dedicated Apache Flink Slack vs. ASF Slack, I'm with Konstantin. I'd expect it to be easier for having more channels and

Re: OOM errors cause by the new KafkaSink API

Hi Martijn, > Have you built your own Flink version? Since Flink doesn't support Scala 2.12.12, the latest Scala version that Flink 1.14 supports is Scala 2.12.7 No, we use the public artifact from here to build our application. How can I kno

Re: Failed to restore from ck, because of KryoException

Hi Liting, Did you ever change your defined class, e.g. xxx.SparkIdentifiers, xxx.ServiceEvent, xxx.Event and so on? Or did you change your flink program? Kryo serializer needs to register class for serialization, which is not friendly for version upgrade. Moreover, except from the compatibil

Re: sharing data between 2 pipelines

why not put the two pipelines together. David Anderson 于2022年5月11日周三 00:13写道： > > This sounds like it might be a use case for something like a > KeyedCoProcessFunction (or possibly a KeyedBroadcastProcessFunction, > depending on the details). These operators can receive inputs from two > diffe

Re: [State Processor API] unable to load the state of a trigger attached to a session window

Is the library, State Processor API, not widely used and no longer maintained? On Mon, Apr 25, 2022 at 3:16 PM Dongwon Kim wrote: > Can anyone help me with this? > > Thanks in advance, > > On Tue, Apr 19, 2022 at 4:28 PM Dongwon Kim wrote: > >> Hi, >> >> I'm using Flink-1.14.4 and failed to loa

Re: Migrating Flink apps across cloud with state

Hi there, to me the simplest and most reliable solution still seems to be to split the stream based on event time. It requires a bit of preparation and assume that you can tolerate some downtime when doing the migration. 1) For Cloud1 you chain a filter to your sources that filters out any record

Re: flink Job is throwing depdnecy issue when submitted to clusrer

Hi there, are you using any of Flink S3 Filesystems? If so, where do you load it from a) lib/ b) plugins/ c) bundled with your Job in a fat JAR b) would be the right way to do it in Flink 1.13. I don't know if this fixes the issue, but IIRC because we introduced the plugin mechansim we don't rel

Re: sharing data between 2 pipelines

This sounds like it might be a use case for something like a KeyedCoProcessFunction (or possibly a KeyedBroadcastProcessFunction, depending on the details). These operators can receive inputs from two different sources, and share state between them. The rides and fares exercise [1] from the flink-

RE: Practical guidance with Scala and Flink >= 1.15

… just for my understanding From the announcements I only got that scala remains only a dependency in the JARs that relate to the Scala API. I never read about plans to drop the Scala API altogether … is that the case?? That would be very unfortunate … What is the state of the affair? Best rega

RE: Notify on 0 events in a Tumbling Event Time Window

Hi Shilpa, There is no need to have artificial messages in the input kafka topic (and I don’t see where Andrew suggests this 😊 ) However your use case is not 100% clear as to for which keys you want to emit 0-count window results , either: * A) For all keys your job has ever seen (that’s e

Converting from table to stream, following Avro schema

Hello, I am consuming Kafka messages with Table API connector. I cannot use DataStream API because it does not support Confluent Avro. After consuming the messages, I am converting to DataStream API and using ProcessFunction. The ProcessFunction makes async http calls and emits results with a com

Re: trigger once (batch job with streaming semantics)

Hi Georg, I'm not aware of those examples being available publicly. Best regards, Martijn On Mon, 9 May 2022 at 23:04, Georg Heiler wrote: > Hi Martijn, > > many thanks for this clarification. Do you know of any example somewhere > which would showcase such an approach? > > Best, > Georg > >

Re: OOM errors cause by the new KafkaSink API

Hi Hua Wei, Have you built your own Flink version? Since Flink doesn't support Scala 2.12.12, the latest Scala version that Flink 1.14 supports is Scala 2.12.7. > Because the new Kafka API needs the new serializer (KafkaRecordSerializationSchema) and seems like cannot use the old one (KafkaSerial

sharing data between 2 pipelines

Hi all i have 2 pipelines: A. receives information from kafka and "holds" that info B. a pipeline which is triggered by a scheduler and every x minutes should send the info i received in pipeline A to another kafka topic As i understood i cannot use the flink state for this since these are differe

Re: table.local-time-zone not working

I guess the ‘values’ you means is the underlying instant value fo TIMESTAMP_LTZ datatype, they are the epoch times which are same in different timezones, that’s the epoch semantics. The session timezone does not affect the underlying value, it only affects the display string in a session, you c

Re: How to get flink to use POJO serializer when enum is present in POJO class

Is this your whole Rule class? If so, then the issue is not the type of the field but that you haven't added a setter and getter for it. On Tue, May 10, 2022 at 7:23 AM Tejas B wrote: > Hi, > I am trying to get flink schema evolution to work for me using POJO > serializer. But I found out that i

Re: [Discuss] Creating an Apache Flink slack workspace

Thanks for starting this discussion again. I am pretty much with Timo here. Slack or Discourse as an alternative for the user community, and mailing list for the contributing, design discussion, etc. I definitely see the friction of joining a mailing list and understand if users are intimidated. I

Re: [Discuss] Creating an Apache Flink slack workspace

I also think that a real-time channel is long overdue. The Flink community in China has shown that such a platform can be useful for improving the collaboration within the community. The DingTalk channel of 10k+ users collectively helping each other is great to see. It could also reduce the bur

Re: table.local-time-zone not working

I have also tried to set it in flink-conf.yaml, it is visible in Job Manager Configurations but still NOW() and PROCTIME() are returning values in UTC. Thanks and Regards , Surendra Lalwani On Tue, May 10, 2022 at 1:09 PM Surendra Lalwani wrote: > Hi Leonard, > > Flink Version is 1.13.6 and I

Re: [Discuss] Creating an Apache Flink slack workspace

Thank you @Xintong Song for sharing the experience of the Flink China community. I'm become convinced we should give Slack a try, both for discussions among the core developers, and as a place where the community can reach out for help. I am in favor of using the ASF slack, as we will need a paid

Re: table.local-time-zone not working

Hi Leonard, Flink Version is 1.13.6 and I am adding property as follows: Configuration tconf = tenv.getConfig().getConfiguration(); tconf.setString("table.local-time-zone", "America/Los_Angeles"); tenv.getConfig().setLocalTimeZone(ZoneId.*of*("America/Los_Angeles")); Thanks and Regards , Suren

Re: table.local-time-zone not working

Hi, Lalwani Could you share how you set this property and your Flink version? Best, Leonard > 2022年5月10日 下午3:01，Surendra Lalwani 写道： > > Hi Team, > > I have tried using this property > table.local-time-zone > > But it seems like it is not making any impact, on calling functions like > PRO

table.local-time-zone not working

Hi Team, I have tried using this property table.local-time-zone But it seems like it is not making any impact, on calling functions like PROCTIME() and NOW(), it always returns UTC value. Any help would be appreciated. Thanks and Regards , Surendra Lalwani -- IMPORTANT NOTICE: This e-mail, inc