Re: Scala Case Class Serialization

Hi, I guess the problematic line where the kryo fallback is happening is here: lazy val myState: MapState[String, TestCaseClass] = getRuntimeContext.getMapState( new MapStateDescriptor[String, TestCaseClass]("test-state", classOf[String], ttestclass.getTypeClass) MapStateDescriptor has

Re: Alternatives of KafkaDeserializationSchema.isEndOfStream()

https://github.com/apache/flink/blob/release-1.13.1/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java#L69 On Wed, Dec 8, 2021 at 12:40 PM Robert Metzger wrote: > Hi Ayush, > > I couldn't find t

Re: Alternatives of KafkaDeserializationSchema.isEndOfStream()

Hi Ayush, I couldn't find the documentation you've mentioned. Can you send me a link to it? On Tue, Dec 7, 2021 at 10:40 AM Ayush Chauhan wrote: > Hi, > > Can you please let me know the alternatives of isEndOfStream() as now > according to docs this method will no longer be used to determine th

Re: Customize Kafka client (module.yaml)

Hi Jérémy, In my understanding of the StateFun docs, you need to pass custom properties using "ingress.spec.properties". For example: ingresses: - ingress: meta: type: io.statefun.kafka/ingress id: project.A/input spec: properties: max.request.size

Re: Weird operator ID check restore from checkpoint fails

Thanks! On Tue, Dec 7, 2021, 22:55 Robert Metzger wrote: > 811d3b279c8b26ed99ff0883b7630242 is the operator id. > If I'm not mistaken, running the job graph generation (e.g. the main > method) in DEBUG log level will show you all the IDs generated. This should > help you map this ID to your code

Re: Which issue or commit should i merge in flink-1.13.3 for buffer debloating?

Hi, I guess all the commits mentioned in all the subtasks of this ticket will give you the feature: https://issues.apache.org/jira/browse/FLINK-23451 Hower, I'm pretty sure that you can't just cherry-pick such a big feature to an older Flink version. I would rather try to fix the connector to up

Re: Weird operator ID check restore from checkpoint fails

811d3b279c8b26ed99ff0883b7630242 is the operator id. If I'm not mistaken, running the job graph generation (e.g. the main method) in DEBUG log level will show you all the IDs generated. This should help you map this ID to your code. On Wed, Dec 8, 2021 at 7:52 AM Dan Hill wrote: > Nothing change

Re: Job Listener not working as expected

Hi Puneet, Are you submitting the Flink jobs using the "/bin/flink" command line tool to a cluster in session mode? Maybe the command line tool is just "fire and forget" submitting the job to the cluster, that's why the listeners are firing immediately. Can you try to use "env.executeAsync()" inst

Re: Weird operator ID check restore from checkpoint fails

Nothing changed (as far as I know). It's the same binary and the same args. It's Flink v1.12.3. I'm going to switch away from auto-gen uids and see if that helps. The job randomly started failing to checkpoint. I cancelled the job and started it from the last successful checkpoint. I'm confus

Re: Weird operator ID check restore from checkpoint fails

Hi Dan, When restoring a savepoint/checkpoint, Flink is matching the state for the operators based on the uuid of the operator. The exception says that there is some state that doesn't match any operator. So from Flink's perspective, the operator is gone. Here is more information: https://nightlie

Re: Weird operator ID check restore from checkpoint fails

I'm restoring the job with the same binary and same flags/args. On Tue, Dec 7, 2021 at 8:48 PM Dan Hill wrote: > Hi. I noticed this warning has "operator > 811d3b279c8b26ed99ff0883b7630242" in it. I assume this should be an > operator uid or name. It looks like something else. What is it? I

Weird operator ID check restore from checkpoint fails

Hi. I noticed this warning has "operator 811d3b279c8b26ed99ff0883b7630242" in it. I assume this should be an operator uid or name. It looks like something else. What is it? Is something corrupted? org.apache.flink.runtime.client.JobInitializationException: Could not instantiate JobManager.

Re: How to write from Flink to a write throttled database?

Write throttled but I may not have data loss, right? I saw this line in producer: https://github.com/klarna-incubator/flink-connector-dynamodb/blob/10e8abaec2ef5473618efba41c31484dadb9ad39/src/main/java/com/klarna/flink/connectors/dynamodb/FlinkDynamoDBSink.java#L202 On Tue, Dec 7, 2021 at 7:4

Job Listener not working as expected

Hi, I have registered a job listener which notifies slack with JobId on successful submission. Also it notifies slack on successful/failed Execution. Now this job listener is working as expected when running on local IDE , but somehow behaving unexpectedly when running on a cluster i.e. both o

Re: How to write from Flink to a write throttled database?

Hi Caizhi, Here is my current configuration: val dynamoDBSinkConfig: DynamoDBSinkConfig = (new DynamoDBSinkConfig.Builder).batchSize(50).queueLimit(20).build() new FlinkDynamoDBSink[Row]( dynamoDBBuilder, "tablename", dynamoDBSinkConfig, mapper ) I think this is batch write. On T

Which issue or commit should i merge in flink-1.13.3 for buffer debloating?

Hi community! Because of the limitation of connector, i couldn't upgrade apache flink from version 1.13.3 to versin 1.14.0. But i really need the important feature of buffer debloating in 1.14.0 for heavy checkpoint at backpressure. So which issue or commit should i merge in flink-1.13.3 for

Re: How to write from Flink to a write throttled database?

Hi! Thanks for the clarification. I'm not familiar with DynamoDB and you might want to modify this connector a bit. Will a WriteRequest immediately send write requests to the database? If yes you may want to instead cache the requests in memory and send them only at snapshots. See [1] for the cod

Re: Issue with incremental checkpointing size

Hi! Thanks for the clarification. Could you expand checkpoint #195 and #196 for details? Slow end-to-end checkpoint time may be caused by various reasons, for example if the data processing is slow it will backpressure the checkpoint (if that is the case, alignment duration should be high for som

Re: How to write from Flink to a write throttled database?

Hi Cazhi, Thanks for your reply! The database is DynamoDB. The connector I use is https://github.com/klarna-incubator/flink-connector-dynamodb. My source is a continuous event stream. My Flink version is 1.12. Best, Jing On Tue, Dec 7, 2021 at 6:15 PM Caizhi Weng wrote: > Hi! > > Which databas

Re: Running a flink job with Yarn per-job mode from application code.

Hi! So you would like to submit a yarn job with Java code, not using /bin/flink run? If that is the case, you'll need to set 'execution.target' config option to 'yarn-per-job'. Set this in the configuration of ExecutionEnvironment and execute the job with Flink API as normal. Kamil ty 于2021年12月

Re: How to write from Flink to a write throttled database?

Hi! Which database are you referring to? If there is no officially supported connector of this database you can create your own sink operator with GenericWriteAheadSink. Note that if you're using Flink < 1.14 and if your source is bounded (that is to say, your source will eventually come to an en

Re: Creating custom connector lib - dependency scope

They should be set to provided so that they are not bundled into the user-jar. You can also take a look at the connectors in the Flink repo to see how they handle dependencies. On 07/12/2021 22:31, Krzysztof Chmielewski wrote: Hi all, I was wondering, when implementing a custom Flink Connect

How to write from Flink to a write throttled database?

Hi, community I have a Kafka stream and want to use Flink for 10 minutes aggregation. However, the number of events is large, and the writes are throttled for the output database for a few seconds during an hour. I was thinking to write from Flink to another Kafka stream and using another Flink ap

Creating custom connector lib - dependency scope

Hi all, I was wondering, when implementing a custom Flink Connector that will be later used as a dependency for other projects, whether dependencies to Flink like: flink-core flink-runtime flink-table-common flink-table-api-java-bridge flink-java etc... Should be in scope provided in the connector

Re: Scala Case Class Serialization

Thanks for quick response. Please find attached a minimal example illustrating the issue. I've added implicit TypeInformation, and checked that I'm importing the scala variant only. Matthias: Just my superficial impression from [1]. Will look into TypeInfoFactory. Thanks again! package com.mystu

Re: Windows and data loss.

Hi, same here there's no pressure... So my overall architecture is as follows. I try to follow the 12 factor app design principles. 1- My APIs all or 99% run inside DC/OS as docker containers. 2- The apps log to the console, which get written to disk by DC/OS log driver. 3- The apps are balanced b

Re: use of Scala versions >= 2.13 in Flink 1.15

Indeed, if you use a scala-free Flink then Scala types would currently go through Kryo, hence why we will recommend to use Java types /for the time being/. We are aware that this is an annoying limitation, and it is certainly not a state we want to at in the long-term. There are some ideas float

Customize Kafka client (module.yaml)

Hi All, I encounter a blocking problem linked to exchanging messages between Stateful functions. The context is: I am sending a very large payload from a Stateful Function to a Kafka topic. I am blocked by the Kafka client (I think) because here is the output of the statefun-manager container:

Re: use of Scala versions >= 2.13 in Flink 1.15

Hi, I guess using scala 2.13 with scala-free Flink 1.15 assumes that it will always use generic/Kryo serialization, which has a large performance penalty (YMMV, but it happens all the time with us when we accidentaly use flink java apis with scala case classes). As far as I know, Flink's set o

Re: Scala Case Class Serialization

Hi Lars, can you please show a small reproducer of the way you construct the DataStream, and which imports do you use? We also often experience similar performance issues with scala, but usually they are related to accidental usage of Flink Java API. A couple of hints from my experience: 1. Ma

Re: Scala Case Class Serialization

Hi Lars, not sure about the out-of-the-box support for case classes with primitive member types (could you refer to the section which made you conclude this?). I haven't used Scala with Flink, yet. So maybe, others can give more context. But have you looked into using the TypeInfoFactory to define

Running a flink job with Yarn per-job mode from application code.

Hello all, I'm looking for a way to submit a Yarn job from another flink jobs application code. I can see that you can access a cluster and submit jobs with a RestClusterClient, but it seems a Yarn per-job mode is not supported with it. Any suggestions would be appreciated. Best Regards Kamil

Alternatives of KafkaDeserializationSchema.isEndOfStream()

Hi, Can you please let me know the alternatives of isEndOfStream() as now according to docs this method will no longer be used to determine the end of the stream. -- Ayush Chauhan Data Platform [image: mobile-icon] +91 9990747111 -- This email is intended only for the person or

Re: Unable to create new native thread error

Hi David, In that case, I will start working on using the CLI instead of the REST API right away. Will update you when I finish. Thanks for the help, Ilan. From: David Morávek Date: Monday, 6 December 2021 at 10:34 To: Ilan Huchansky Cc: user@flink.apache.org , Start.io SDP Subject: Re: Un

Scala Case Class Serialization

Hello, We're running Flink 1.14 with scala, and we're suspecting that performance is suffering due to serialization of some scala case classes. Specifically we're seeing that our Case Class "cannot be used as a POJO type because not all fields are valid POJO fields, and must be processed as Generic

Re: Re: Re: how to run streaming process after batch process is completed?

Hello, One of the applications Spire [1] is using Flink for is to process AIS [2] data collected by our satellites and from other sources. AIS is transmitting a ships' static and dynamic information, such as names, callsigns or positions. One of the challenges processing AIS data is that there are

Re: use of Scala versions >= 2.13 in Flink 1.15

We haven't changed anything significant in 1.14. Whether the 2.13 job can run on Scala 2.12 depends a bit on the job (and of course, used libraries!); it depends on the backwards-compatibility from Scala, which APIs are used and what kind of Scala magic is being employed. We haven't really te

Re: [DISCUSS] Drop Zookeeper 3.4

Since this is only relevant for 1.15, if you intend to migrate to 1.15 close to the release, then somewhere around February. The only resource I could find for migrating Zookeeper is this FAQ: https://cwiki.apache.org/confluence/display/ZOOKEEPER/Upgrade+FAQ On 07/12/2021 04:02, Dongwon Kim w

Re: Unsubscribe

Please send an empty email to: user-unsubscr...@flink.apache.org to unsubscribe yourself from the list. On Tue, Dec 7, 2021 at 4:25 PM wrote: > Unsubscribe > > >

Re: use of Scala versions >= 2.13 in Flink 1.15

Hi Chesnay, thanks for the info - this is really good news for us. I set up a playground using the snapshot from yesterday [1] and a really quick and short Job using Scala 2.13 [2] The job starts and returns correct results. Even the use of a case class against the Java API is possible. Th

Unsubscribe

Unsubscribe