Why should we use an evictor operator in flink window

Hi, all When using Flink window, I do not know why should we use an evictor operator in flink window, after reading the EvictingWindowOperator.java, I think we could do all the [evcitor] things in the userfunction. Could anyone help me to understand these? I have digged the history of

Re: Order of events in a Keyed Stream

Hi, Maybe the messages of the same key should be in the *same partition* of Kafka topic 2018-07-29 11:01 GMT+08:00 Hequn Cheng : > Hi harshvardhan, > If 1.the messages exist on the same topic and 2.there are no rebalance and > 3.keyby on the same field with same value, the answer is yes. > >

Re: Are savepoints / checkpoints co-ordinated?

AFAIK, "Cancel Job with Savepoint" will stop checkpointScheduler --> trigger a savepoint, then cancel your job. there will no more checkpoints. 于2018年10月12日周五 上午1:30写道： > Hi, > > > > I had a couple questions about savepoints / checkpoints > > > > When I issue "Cancel Job with Savepoint", how is

Re: How do I initialize the window state on first run?

IIUC, we can't initialize state at first run, maybe you could store the aggregated data in another place other than use flink's state, then use flink to aggregate the data realtime. bupt_ljy 于2018年10月12日周五 下午3:33写道： > Hi, vivo, > > My Flink program is to aggregate the data of a whole day,

Re: Is there a better way to debug ClassNotFoundException from FlinkUserCodeClassLoaders?

Hi, Hao Sun For debugging the `ClassNotFoundException`, maybe the Arthas[1] tool can help. [1] Arthas : https://github.com/alibaba/arthas Hao Sun 于2019年1月3日周四 下午10:08写道： > > I am on Flink 1.7.1 and K8S. > I said "suddenly" because my program worked fine

Re: Change Window Size during runtime

Hi, Rad We can not change the window size during runtime as Chesnay said. Could you please share the reason why would you want to change the window size during runtime? Rad Rad 于2019年1月3日周四 上午8:03写道： > Hi All, > > I have one stream is consumed by FlinkKafkaConsumer which will be joined > with

Re: Streaming Checkpoint - Could not materialize checkpoint Exception

Hi, Sohi You can check out doc[1][2] to find out the answer. [1] https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/stream/state/checkpointing.html#enabling-and-configuring-checkpointing [2] https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/restart_strategies.html

Re: Streaming Checkpoint - Could not materialize checkpoint Exception

Hi, Sohi Seems like the checkpoint file `hdfs:/pipeline/job/checkpoints/e9a08c0661a6c31b5af540cf352e1265/chk-470/5fb3a899-8c0f-45f6-a847-42cbb71e6d19` did not exist for some reason, you can check the life cycle of this file from hdfs audit log and find out why the file did not exist. maybe the

Re: Check-pointing error

Hi,Felipe Quirce Could you reproduce this in stand alone mode? or could you share your code? Best Congxian Felipe Quirce 于2018年11月30日周五 下午9:34写道： > Hi Chesnay, > > I tried with the version 1.7.0 and I had the same error. > > 2018-11-30 13:13:00,718 INFO

Re: Multiple MapState vs single nested MapState in stateful Operator

Hi, Gagan Agrawal In my opinion, I prefer the first. Here is the reason. In RocksDB StateBackend, we will serialize the key, namespace, user-key into a serialized bytes (key-bytes) and serialize user-value to serialized bytes(value-bytes) then insert into the key-bytes/value-bytes into

Re: What happen to state in Flink Task Manager when crash?

Hi, Siew Wai Yow When the job is running, the states are stored in the local RocksDB, Flink will copy all the needed states to checkpointPath when doing a Checkpoint. If there have any failures, the job will be restored from the last previously *Successfully* checkpoint, and assign the restored

Re: An exception about checkpoint

Hi, Leor Seems like the checkpoint file `/app/flink/checkpoints/918f49d9eb23bdea016e865182718a06/chk-10631/6e86cce6-9ac9-4856-b0a3-42c81fcafadc` did not exist for some reason, you can check the life cycle of this file from hdfs audit log and find out why the file did not exist. maybe the

Re: Reply: Re: What happen to state in Flink Task Manager when crash?

Hi, Siew Wai Yow Yes, David is correct, the TM must be recovered, the number of TMs before and after the crash must be the same. In my last reply, I want to say that the states may not on the same TM after the crash. Sorry for the unclear description. Siew Wai Yow 于2019年1月12日周六 下午6:44写道： >

Re: Reply: Re: Reply: Re: What happen to state in Flink Task Manager when crash?

Hi, Yow I think there is another restart strategy in flink: region failover[1], but I could not find the documentation, maybe someone else may help here, For region failover, please take a look at this issue[2] before you use it. And you can take a look at this FLIP[3]. [1]

Re: How test and validate a data stream software?

Hi, Alexandre Maybe the blog post[1] can be helpful. [1] https://www.da-platform.com/blog/extending-the-yahoo-streaming-benchmark Alexandre Strapacao Guedes Vianna 于2019年1月23日周三 下午9:54写道： > Hello People, > > I'm conducting a study for my PhD about applications using data stream > processing,

Re: [Flink 1.6] How to get current total number of processed events

Hi, Nhan Do you want the total number of the current parallelism or the operator? If you want the total number of the current parallelism, Is the operator state[1] satisfied with your use case? [1]

Re: kafka corrupt record exception

Hi As you said, consume from ubuntu terminal has the same error, maybe you could send a email to kafka user maillist. Best, Congxian On Apr 1, 2019, 05:26 +0800, Sushant Sawant , wrote: > Hi team, > I am facing this exception, > org.apache.kafka.common.KafkaException: Received exception when

Re: Partitioning key range

Hi Davood Maybe a custom KeySelector can be helpful, you can define the key used to partition the stream. You can ref the code[1] for detail. [1] 

Re: How to add unit test for flink sql ?

Hi, Lifei Maybe org.apache.flink.table.runtime.stream.sql.JavaSqlITCase can be helpful. Best, Congxian Lifei Chen 于2019年2月27日周三 下午4:20写道： > Hi, all: > > I finished a flink streaming job with flink sql, which read data from > kafka and write bach to elasticsearch. > > I have no idea how to

Re: How to debug difference between Kinesis and Kafka

Hi Stephen If the window has not been triggered ever, maybe you could investigate the watermark, maybe the doc[1][2] can be helpful. [1]  https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/stream/operators/windows.html#interaction-of-watermarks-and-windows [2] 

Re: Externalised checkpoint keeps ValueState after a crash of a Flink cluster

hi, Min Complete Checkpoint contains the snapshot of all states, and when recovery from checkpoint, all the states will be recovered from checkpoint, from what you described, I guess when the job manager gets killed, there is an onging but not completed checkpoint. Maybe the doc[1] can be

Re: KeyBy distribution across taskslots

Hi Maybe you could add a prefix for each key, so the hot keys can distributed to many tasks. Best, Congxian On Feb 28, 2019, 21:16 +0800, Yun Tang , wrote: > Hi, > > If you noticed that some key groups are hot and in high load, you could try > to increase the total key groups number (by

Re: How do I compute the average and keep track of a state over a window in DataStream?

Hi Felipe Maybe you could use process function[1] [1]  https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/stream/operators/process_function.html Best, Congxian On Feb 28, 2019, 22:47 +0800, Felipe Gutierrez , wrote: > Hi all, > > I want to compute the average of two stream data

Re: Conflicting Cassandra versions while using flink-connector-cassandra

Hi Gustavo Momenté If you want the both driver versions coexist, maybe you could try maven shade plugin[1] [1] https://maven.apache.org/plugins/maven-shade-plugin/ Best, Congxian On Mar 13, 2019, 02:22 +0800, Gustavo Momenté , wrote: > I'm having trouble using CassandraSink while also using 

Re: Rocksdb to filesystem state migration errors

Hi Lakshmi Currently, we can’t switch between rocksdb and filesystem backend using savepoint, there is an issue to fix this[1]. [1] https://issues.apache.org/jira/browse/FLINK-11254 Best, Congxian Lakshmi Gururaja Rao 于2019年3月15日周五 上午8:07写道： > Hey all, > > I'm trying to do a state

Re: MapState - TypeSerializer

Hi, Alexey In your case, only TypeSerializer2 will be stored in meta information. and TypeSerializer2 and TypeSeriaizer1 have to be compatible. Best, Congxian Alexey Trenikhun 于2019年2月8日周五 上午10:39写道： > What if I’m using RocksDB, and MapState had single entry and > TypeSerializer1, then we

Re: Window elements for certain period for delayed processing

Hi, simpleusr Maybe custom trigger[1] can be helpful. [1]  https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/stream/operators/windows.html#triggers Best, Congxian On Feb 15, 2019, 13:15 +0800, simpleusr , wrote: > Hi, > > My ultimate requirement is to stop processing of certain

Re: Incorrect Javadoc in CheckpointedFunction.java?

Hi Chirag I think the doc is outdated, the comments in CheckpointFuncion.java on master now[1] is `get the state data structure for the per-partition state` [1] 

Re: late element and expired state

Hi, Aggarwal. Your strategy to limit the total state is right. And there did not have an API will give user the hint about cleared/expired state. I think you can associate every key with two states: one as the ttl-state, one as the “seen”-state(ValeState)—the “seen”-state will tell you

Re: [ANNOUNCE] New Flink PMC member Thomas Weise

Congrats Thomas! Best, Congxian françois lacombe 于2019年2月13日周三 下午4:52写道： > Congratulation Thomas > > Thanks for help you provide and useful inputs > > François > > Le mer. 13 févr. 2019 à 03:13, Kurt Young a écrit : > >> Congrats Thomas! >> >> Best, >> Kurt >> >> >> On Wed, Feb 13, 2019 at

Re: flink restoring from state

Hi, Avi I think the "*Checkpoint failed: The assigned slot container_e02_1550091678485_0001_01_23_7 was removed"*(this may be a container failure or something else, could double check the taskamanger log for more information)and *"**Checkpoint failed: Checkpoint Coordinator is suspending"

Re: About KafkaConsumer and WM'ing and EventTime charactersitics

Hi Vishal May this doc[1] be helpful for you. [1] https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/connectors/kafka.html#kafka-consumers-and-timestamp-extractionwatermark-emission Best, Congxian Vishal Santoshi 于2019年1月30日周三 上午4:36写道： > It seems from >

Re: [Flink 1.6] How to get current total number of processed events

llel > operator instance. > Indeed, I expect to get the total number of all parallel operator > instances. > > Is there a way to sum up all these operator states , please? > > Best regard, > Nhan > > > > *De :* Congxian Qiu [mailto:qcx978132...@gmail.com] > *Envoyé

Re: Videos and slides on Flink Forward Beijing

Hi Paul The video is available here [1]. [1] https://m.bilibili.com/space/33807709 Paul Lam 于2019年2月1日 周五15:18写道： > Hi, > > It’s been a while since Flink Forward Beijing, would the videos and slides > be available on the website? Thanks! > > Best, > Paul Lam > > -- Best, Congxian

Re: Reverse of KeyBy

Hi Aggarwal How about keyBy(LargeMessageID) first, then assemble these fragments back into LargeMessages, then keyBy(MeyKey)? Best, Congxian Aggarwal, Ajay 于2019年2月2日周六 上午5:42写道： > I am new to Flink. I am trying to figure out if there is an operator that > provides reverse functionality of

Re: KeyBy is not creating different keyed streams for different keys

Hi Harshith You can replace the GenericDataObject with Tuple3 and keyBy("A", "B") with keyBy(1, 2) then have a try. And you can see the doc[1] for reference also. [1] https://ci.apache.org/projects/flink/flink-docs-stable/dev/api_concepts.html#define-keys-using-key-selector-functions Best,

Re: NullPointerException on StreamTask

Hi From the code[1] said, seems `chainedConfigs.get(edge.getSourceId())` is null(chainedConfigs will never be null and edge would not be null), but I can’t tell why this can happen, maybe some one else can help here. [1] 

Re: Controlling the amount of checkpoint files

teBackend("file:///flink/checkpoints", > true) > env.setStateBackend(checkpointingBackend) > > > > Boris Lublinsky > FDP Architect > boris.lublin...@lightbend.com > https://www.lightbend.com/ > > On Jun 10, 2019, at 1:07 AM, Congxian Qiu wrote: > > Hi

Re: How to join/group 2 streams by key?

Hi John I've seen other people have the same problem to solve, the following is their solution: union the two Datastreams, then use ProcsssFunction[1] to solve this, will also register timers to do GC things. [1]

Re: Flink state: complex value state pojos vs explicitly managed fields

Hi, If you use RocksDBStateBackend, one member one state will get better performance. Because RocksDBStateBackend needs to de/serialize the key/value when put/get, with one POJO value, you need to de/serializer the whole POJO value when put/get. Best, Congxian Timothy Victor 于2019年6月17日周一

Re: [External] checkpoint metadata not in configured directory state.checkpoints.dir

Hi, Vishal If you want to restart from the last competed external checkpoint of the previous stoped job, you need to track the checkpoint path and restart from it. [1] https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/state/checkpoints.html#resuming-from-a-retained-checkpoint Best,

Re: [External] checkpoint metadata not in configured directory state.checkpoints.dir

t; regarding this ? > > Thanks, > Vishal Sharma > > On Thu, Jun 20, 2019 at 11:17 AM Congxian Qiu > wrote: > >> Hi, Vishal >> If you want to restart from the last competed external checkpoint of the >> previous stoped job, you need to track the checkpoint path and res

Re: Checkpoint expired before completing with cleanupInRocksdbCompactFilter

Hi, Mu Is there anything  looks like `Received  late message for now expired checkpoint attempt ${checkpointID} from ${taskkExecutionID} of job ${jobID}` in JM log? If yes, that means this task complete the checkpoint too long (maybe receive barrier too late, maybe spend too much time to do

Re: Checkpoints periodically fail with hdfs as the state backend - Could not flush and close the file system output stream

Hi Pedro From the previous given log, I found that checkpoint 65912 has been expired then, the raise the IOException. When some checkpoint expired, the checkpoint dir will be deleted(CheckpointCoordinator#549 on release-1.6 branch), and the unfinished task will still write to the previous

Re: Count Window Trigger that only fires once

Hi Frank, Seems you want a custom trigger, maybe the doc[1] can help. [1]  https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/stream/operators/windows.html#triggers Best Congxian On May 23, 2019, 23:38 +0800, Frank Wilson , wrote: > Hi, > > Is there a way to make the count window

Re: Propagating delta from window upon trigger

Hi, Nikhil Window will emit all state to downstream. Can you clear the state while triggering? Nikhil Goyal 于2019年5月19日 周日01:03写道： > I have window of 1 hour and trigger of 5 min. I want to know if every 5 > min Flink is writing the entire window or only the keys which changed. > > On Sat, May

Re: Checkpoints periodically fail with hdfs as the state backend - Could not flush and close the file system output stream

Hi Pedro Could you please share the audit log for file `/flink/data/checkpoints/76f7b4f5c679e8f2d822c9c3c73faf5d/chk-65912/68776faf-b687-403b-ba0c-17419f8684dc`, seems this did not exist cause this problem (maybe this file was created and deleted for some reason) Best, Congxian Andrey Zagrebin

Re: How to export all not-null keyed ValueState

Hi, Averell AFAIK, we can't get all the key-values from value state, but MapState has a function called `entries` can do this, maybe can use MapState as a workaround. On May 7, 2019, 16:16 +0800, Averell , wrote: > Hi, > > I have a keyed value state which is available for only about 1% the total

Re: RocksDB backend with deferred writes?

Hi, David When you flush data to db, you can reference the serialize logic[1], and store the serialized bytes to RocksDB. [1] 

Re: Checkpointing and save pointing

Hi, Boris TM will also need to write to the external volume. Best, Congxian On May 8, 2019, 03:56 +0800, Boris Lublinsky , wrote: > I am planning to use external volume for this. My understanding is that it > needs to be mounted only to the job manager, not the task managers. Is this >

Re: [ANNOUNCE] Apache Flink 1.8.1 released

Thanks for being the release manager and the great job Best, Congxian Jark Wu 于2019年7月3日周三 上午10:23写道： > Thanks for being the release manager and the great job! > > Cheers, > Jark > > On Wed, 3 Jul 2019 at 10:16, Dian Fu wrote: > > > Awesome! Thanks a lot for being the release manager. Great

Re: [ANNOUNCE] Rong Rong becomes a Flink committer

Congratulations Rong! Best, Congxian Biao Liu 于2019年7月12日周五 下午4:40写道： > Congrats, Rong! > > > Hequn Cheng 于2019年7月12日周五 下午1:09写道： > >> Congratulations Rong! >> >> Best, Hequn >> >> On Fri, Jul 12, 2019 at 12:19 PM Jeff Zhang wrote: >> >>> Congrats, Rong! >>> >>> >>> vino yang 于2019年7月12日周五

Re: HDFS checkpoints for rocksDB state backend:

Hi Andrea As the NoClassDefFoundError, could you please verify that there exist `org.apache.hadoop.hdfs.protocol.HdfsConstants*` *in your jar. Or could you use Arthas[1] to check if there exists the class when running the job? [1] https://github.com/alibaba/arthas Best, Congxian Andrea Spina

Re: Error restoring from checkpoint on Flink 1.8

Hi, >From the given error message, this seems flink can't open RocksDB because of the number of column family mismatch, do you mind sharing a minimum job which can reproduce this problem? Best, Congxian Ning Shi 于2019年4月21日周日 上午10:56写道： > For clarification, one of the operators in the chain

Re: Missing state in RocksDB checkpoints

Hi, Ning From the log message you given, the two operate share the same directory, and when snapshot, the directory will be deleted first if it exists(RocksIncrementalSnapshotStrategy#prepareLocalSnapshotDirectory). I did not find an issue for this problem, and I don’t thinks this is a problem

Re: Missing state in RocksDB checkpoints

relevant information. > > Thanks, > > — > Ning > > > On Apr 23, 2019, at 2:54 AM, Congxian Qiu wrote: > > > > From the log message you given, the two operate share the same directory, > > and when snapshot, the directory will be deleted first if it

Re: Watermark for each key?

There was someone working in IoT asking me whether Flink supports per-key watermark also. I’m not sure if we can do the statistics by using raw state manipulating. We create a single state for every single key, and when receiving a key, we extract the timestamp and to see if we need to send

Re: Changing the way keys are defined breaks savepoints

Hi, >From what you described, does the key have the same type before and after the change? and for using Lambda expression, maybe the doc[1] can be helpful(especially the limitation) [1] https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/java_lambdas.html#examples-and-limitations

Re: Update Checkpoint and/or Savepoint Timeout of Running Job without restart?

Hi Currently, we can't change a running job's checkpoint timeout, but there is an issue[1] which wants to set a separate timeout for savepoint. [1] https://issues.apache.org/jira/browse/FLINK-9465 Best, Congxian Aaron Levin 于2019年8月17日周六 上午12:37写道： > Hello, > > Question: Is it possible to

Re: Externalized checkpoints

Hi, Vishwas As Zhu Zhu said, you can set "state.checkpoints.num-retained"[1] to specify the maximum number of completed checkpoints to retain. maybe you can also ref the external checkpoint cleanup type[2] config for how to clean up the retained checkpoint[2] [1]

Re: [ANNOUNCE] Apache Flink 1.9.0 released

Congratulations, and thanks for everyone who make this release possible. Best, Congxian Kurt Young 于2019年8月23日周五 上午8:13写道： > Great to hear! Thanks Gordon for driving the release, and it's been a > great pleasure to work with you as release managers for the last couple of > weeks. And thanks

Re: Why is the size of each checkpoint increasing?

Hi Andrew >From Flink doc[1], there is "Flink guarantees removal only for time-based windows and not for other types, *e.g.* global windows (see Window Assigners ). ", Seems the state of the

Re: Capping RocksDb memory usage

Hi Maybe FLIP-49[1] "Unified Memory Configuration for TaskExecutors" can give some information here [1] http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-FLIP-49-Unified-Memory-Configuration-for-TaskExecutors-td31436.html Best, Congxian Cam Mach 于2019年8月9日周五 上午4:59写道： >

Re: Re: [ANNOUNCE] Hequn becomes a Flink committer

Congratulations Hequn! Best, Congxian Yu Li 于2019年8月8日周四 下午2:02写道： > Congratulations Hequn! Well deserved! > > Best Regards, > Yu > > > On Thu, 8 Aug 2019 at 03:53, Haibo Sun wrote: > >> Congratulations! >> >> Best, >> Haibo >> >> At 2019-08-08 02:08:21, "Yun Tang" wrote: >>

Re: [DISCUSS] Create a Flink ecosystem website

Robert and Daryl, thanks for the great work, I tried the website and filed some issues on Github. Best, Congxian Robert Metzger 于2019年7月17日周三 下午11:28写道： > Hey all, > > Daryl and I have great news to share. We are about to finish adding the > basic features to the ecosystem page. > We are at a

Re: Checkpoints timing out for no apparent reason

Hi The image did not show. incremental checkpoint includes: 1) flush memtable to sst files; 2) checkpoint of RocksDB; 3) snapshot metadata; 4) upload needed sst files to remote, all the first three steps are in sync part, and the fourth step in async part, could you please check whether the sync

Re: MapSate within Aggregate function

Hi Ahmad AFAIK, Flink currently does not support stores incoming elements to a MapState, maybe the window function[1] can be help [1] https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/stream/operators/windows.html#window-functions Best, Congxian Ahmad Hassan 于2019年7月25日周四

Re: Increasing trend for state size of keyed stream using ProcessWindowFunction with ProcessingTimeSessionWindows

Hi Oliwer, >From the description, Seems the state didn't be cleared, maybe you could check how many {{windowState.clear()}} was triggered in {{WindowOperator#processElement}}, and try to figure it out why the state did not be cleared. Best, Congxian Oliwer Kostera 于2019年9月27日周五 下午4:14写道： >

Re: Broadcast state

Hi, Could you use some cache system such as HBase or Reids to storage this data, and query from the cache if needed? Best, Congxian Navneeth Krishnan 于2019年10月1日周二 上午10:15写道： > Thanks Oytun. The problem with doing that is the same data will be have to > be stored multiple times wasting

Re: Debugging slow/failing checkpoints

Hi Steve 1. Do you use exactly once or at least once? 2. Do you use incremental or not 3. Do you have any timer, and where does the timer stored(Heap or RocksDB), you can ref the config here[1], you can try store the timer in RocksDB. 4. Does the align time too long 5. You can check if it is

Re: Loading state from sink/understanding recovery options

Hi, Eduardo Currently, we can't load state from the outside(there is an ongoing jira[1] to do this), in the other word, if you disable checkpoint, and use the Kafka/database as your state storage, you should do the deduplication things by yourself. Just curious, which state backend do you use,

Re: RocksDB and local file system

Hi The path will store the checkpoints, and Flink will fetch the checkpoint files to restore the state if any failure occurred. If you specify the local file system, when restoring from checkpoint, Flink may can't find the checkpoint files, and can't restore from last checkpoints. Best,

Re: slow checkpoints

Hi Currently, checkpoint may be faile in high back pressure scenario, because the barrier alignment can't be done in expected time, you should fix the back pressure problem first. There is a FLIP[1] that wants to fix this issue. [1]

Re: Keyed raw state - example

Hi Currently, I think you can ref the implementation of timerservice[1] which used the raw keyed state, the snapshot happens in AbstractStreamOperator#snapshotState(), for using Raw State you need to implement a new operator[2]. There is an issue wants to give some example for raw state[2] [1]

Re: Apache Airflow - Question about checkpointing and re-run a job

Hi Yes, checkpoint data locates under jobid dir. you can try to restore from the retained checkpoint[1] [1] https://ci.apache.org/projects/flink/flink-docs-release-1.9/ops/state/checkpoints.html#resuming-from-a-retained-checkpoint Best, Congxian M Singh 于2019年11月18日周一 上午2:54写道： > Folks -

Re: Flink vs Kafka streams

Hi >From your description, seems the big problem is scale in and out, and there maybe a big downtime for trigger savepoint and restore from the savepoint. Previously, we have proposed a feature named stop-with-checkpoint[1] same as the stop-with-savepoint, but triggering a checkpoint instead of

Re: Flink savepoint(checkpoint) recovery dev debug

Hi You can ref CheckpointMetadataLoadingTest#testLoadAndValidateSavepoint for more information. Even more, you can ref the line 136-line145 [1], and init a CompletedCheckpointStorageLocation to make it work. [1]

Re: Cron style for checkpoint

Hi Currently, Flink does not support such feature, from what you describe, does set an appropriate timeout for checkpoint can solve your problem? Best, Congxian shuwen zhou 于2019年11月21日周四 下午12:06写道： > Hi Jiayi, > It would be great if Flink could have a user defined interface for user to >

Re: Cron style for checkpoint

On the other hand, >> savepoint in Flink behaves more like a user control behavior, can savepoint >> not satisfy your demands for crontab? >> >> >> >> Best >> >> Yun Tang >> >> >> >> *From: *Congxian Qiu >> *Date: *

Re: Re:Apache Flink - Operator name and uuid best practices

Mans > > On Wednesday, November 20, 2019, 09:13:39 PM EST, Congxian Qiu < > qcx978132...@gmail.com> wrote: > > > Hi > > Currently, the last part (uuid a4d87cda-2afd-47d4-8d3f-b0658466fb2d) is > generated by UUID.randomUUID(), so there is not a easy way to map this to &g

Re: Savepoints and checkpoints

Hi First, Checkpoint for Flink is a distributed snapshot of the job. As Yun said, Kafka consumer will snapshot the topic name and partition to the checkpoint, then when restoring from the last checkpoint you do not know about the newly topic name. Inner the checkpoint, you can think checkpoint as

Re: Kinesis Connector and Savepoint/Checkpoint restore.

Hi For idle shards, I think restore from the previous not consumed data is ok, because Flink did not consume any data before, but for not idle shards this is problematic. From my experience of other connectors, could you check whether the "error" shards are newly split? maybe the newly split

Re: Interval Join Late Record Metrics

Hi Chris >From the code[1], currently, IntervalJoin will ignore the late data silently, maybe you can create an issue to track this. [1]

Re: Custom Partitioning with keyed state

Hi Have you tried the key selector function[1]? [1] https://ci.apache.org/projects/flink/flink-docs-stable/dev/api_concepts.html#define-keys-using-key-selector-functions Best, Congxian Heidi Hazem Mohamed 于2019年10月27日周日 下午11:04写道： > Hi, > > What I want : I have my own partitioning technique

Re: FLINK-13497 / "Could not create file for checking if truncate works" / HDFS

Hi >From the given stack trace, maybe you could solve the "replication problem" first, File /okd-dev/3fe6b069-43bf-4d86-9762-4f501c9db16e could only be replicated to 0 nodes instead of minReplication (=1). There are 2 datanode(s) running and no node(s) are excluded in this operation, and maybe

Re: FLINK-13497 / "Could not create file for checking if truncate works" / HDFS

ts mentioned in various answers, in particular the one about > temp files. > Thanks > Adrian > > > - Original message - > From: Congxian Qiu > To: Adrian Vasiliu > Cc: user > Subject: [EXTERNAL] Re: FLINK-13497 / "Could not create file for chec

Re: Could not load the native RocksDB library

FYI Maybe this is an env problem. I encountered this problem when running flink 1.9 on k8s, but it was success when running on yarn. did not figure out why this happened, will update here after find it out. Best, Congxian Thad Truman 于2019年10月23日周三 上午1:33写道： > Hi Samya, > > > > Were you able

Re: Broadcast state

slot can read from this shared memory. > > Thanks > > On Wed, Oct 9, 2019 at 12:13 AM Congxian Qiu > wrote: > >> Hi, >> >> After using Redis, why there need to care about eliminate duplicated >> data, if you specify the same key, then Redis will do the deduplicate

Re: standalone flink savepoint restoration

Hi Do you specify the operatorid for all the operators?[1][2], asking this because from the exception you gave, if you only add new operators and all the old operators have specified operatorid, seems there would not throw such exception. [1]

Re: Using MapState clear, put methods in snapshotState within KeyedCoProcessFunction, valid or not?

Hi >From the exception `No key set. This method should not be called outside of a keyed context.` it means that the key current passed in is null. In my opinion, it's something wrong here if there will throw an exception when no data arrive. could you please share the whole stack and a minimal

Re: [DISCUSSION] Using `DataStreamUtils.reinterpretAsKeyedStream` on a pre-partitioned Kafka topic

Hi >From the doc[1], the DataStream MUST already be pre-partitioned in EXACTLY the same way Flink’s keyBy would partition the data in a shuffle w.r.t. key-group assignment. you should make sure that the key locates in the right key-group, and the key-group locates in the right parallelism. you

Re: Apache Flink - Troubleshooting exception while starting the job from savepoint

Hi The problem is that the specified uid did not in the new job. 1. As far as I know, the answer is yes. There are some operators have their own state(such as window state), could you please share the minimal code of your job? 2.*truely* stateless operator do not need to have uid, but for the

Re: Re:Apache Flink - Operator name and uuid best practices

Hi Currently, the last part (uuid a4d87cda-2afd-47d4-8d3f-b0658466fb2d) is generated by UUID.randomUUID(), so there is not a easy way to map this to the assigned in the application. In another word, the last part (uuid a4d87cda-2afd-47d4-8d3f-b0658466fb2d) belongs to one checkpoint, and the

Re: What happens to a Source's Operator State if it stops being initialized and snapshotted? Accidentally exponential?

Hi Do you use UNION state in your scenario, when using UNION state, then JM may encounter OOM because each TDD will contains all the state of all subtasks[1] [1] https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/state.html#using-managed-operator-state Best, Congxian Aaron

Re: ArrayIndexOutOfBoundException on checkpoint creation

Hi Which version are you using now(if on some old version, could you please try if this exception is till there on Flink 1.9), on the other hand, did you try RocksDBStateBackend for this? Best, Congxian Theo Diefenthal 于2019年11月26日周二 下午6:52写道： > Hi, > > We have a pipeline with a custom

Re: How to recover state from savepoint on embedded mode?

Hi, You can recovery from checkpoint/savepoint if JM can read from the given path. no math which mode the job is running on. Best, Congxian Reo Lei 于2019年11月26日周二 下午12:18写道： > > > -- Forwarded message - > 发件人： Reo Lei > Date: 2019年11月26日周二 上午9:53 > Subject: Re: How to

Re: Apache Flink - Troubleshooting exception while starting the job from savepoint

Thanks again for your response. > > Mans > > On Monday, November 25, 2019, 09:24:42 AM EST, Congxian Qiu < > qcx978132...@gmail.com> wrote: > > > Hi > > The problem is that the specified uid did not in the new job. > 1. As far as I know, the answer is yes. There are so

Re: Per Key Grained Watermark Support

Hi There was a discussion about this issue[1], as the previous discussion said at the moment this is not supported out of the box by Flink, I think you can try keyed process function as Lasse said. [1]

Re: Building Flink 1.6.4 fails with "object scala in compiler mirror not found"

Hi if you just want to skip the test, do you try to add `-DskipTests` when executing maven command. Best, Congxian Aikio, Torste 于2019年10月7日周一 下午11:36写道： > Hi, > > I'm trying to build Flink 1.6.4 from source and some of the tests for > flink-scala module are failing for me. Are there some

Re: Group by multiple fields

Hi Miguel Maybe the doc[1] about how to specifying the keys can help. [1] https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/api_concepts.html#specifying-keys Best, Congxian Miguel Farrajota 于2019年10月8日周二 上午12:09写道： > Hi, > > I'm looking to do some result aggregations on a event

  1   2   3   4   5   >