Does flink configuration support configed by environment variables?

Hi guys, I am using flink 1.7.2 deployed by kubernetes, and I want to change the configurations about flink, for example customize `taskmanager.heap.size`. Does flink support using environment variables to override configurations in `conf/flink-conf.yaml` ?

Re: AskTimeoutException - Cannot deploy task

Hi All, Following my previous mail below I see the exception below. I really appreciate any help here Attached is log files Looking at the logs we see this exception all around : 2019-03-28 23:51:58,460 WARN org.apache.kafka.common.network.Selector - Unexpected error from kaf

Re: TUMBLE function with MONTH/YEAR intervals produce milliseconds window unexpectedly

btw the max DAY window that is allowed is 99 days. After that it blows up here: https://github.com/apache/calcite/blob/master/core/src/main/java/org/apache/calcite/sql/SqlIntervalQualifier.java#L371 "SQL validation failed. From line 12, column 19 to line 12, column 36: Interval field value 100 exc

Re: TUMBLE function with MONTH/YEAR intervals produce milliseconds window unexpectedly

Dawid, After the above change my SQL (that uses TUMBLE(rowtime, INTERVAL '1' MONTH)) fails with an error now: *(testing with org.apache.flink:flink-table_2.11:jar:1.7.1:compile now)* org.apache.flink.table.api.TableException: *Only constant window intervals with millisecond resolution are support

Re: Schema Evolution on Dynamic Schema

Hmm kinda stuck here. Seems like SQL Group by is translated to a *GroupAggProcessFunction* which stores a state for every aggregation element (thus flattening the map items for state store). Seems like there's no way around it. Am i wrong? is there any way to evolve the map elements when doing *SEL

AskTimeoutException - Cannot deploy task

Hi, I see the following exceptions, will really appreciate any help on that Thanks Avi This is the first one (out of three) : java.lang.Exception: Cannot deploy task KeyedProcess -> Sink: Unnamed (3/100) (2c9646634afe1488659da404e92697b0) - TaskManager (container_e03_1553795623823_0001_01_000

Re: Calcite SQL Map to Pojo Map

Apparently the solution is to force map creating using UDF and to have the UDF return Types.GENERIC(Map.class) That makes them compatible and treated both as GenericType Thanks! -- Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: TUMBLE function with MONTH/YEAR intervals produce milliseconds window unexpectedly

Thanks Dawid! Can you please point me to a jira which tracked the fix? Thanks! Vinod On Thu, Mar 28, 2019 at 12:46 PM Dawid Wysakowicz wrote: > It should be fixed since version 1.6.3. > Best, > Dawid > > > [1] > https://issues.apache.org/jira/browse/FLINK-11017?jql=project%20%3D%20FLINK%20AND%2

Re: TUMBLE function with MONTH/YEAR intervals produce milliseconds window unexpectedly

Doh! Sorry about that! :) Thanks again! On Thu, Mar 28, 2019 at 12:49 PM Dawid Wysakowicz wrote: > I did ;) but here is the link one more time: > https://issues.apache.org/jira/browse/FLINK-11017?jql=project%20%3D%20FLINK%20AND%20text%20~%20Month > > On Thu, 28 Mar 2019, 20:48 Vinod Mehra, wro

Re: TUMBLE function with MONTH/YEAR intervals produce milliseconds window unexpectedly

I did ;) but here is the link one more time: https://issues.apache.org/jira/browse/FLINK-11017?jql=project%20%3D%20FLINK%20AND%20text%20~%20Month On Thu, 28 Mar 2019, 20:48 Vinod Mehra, wrote: > Thanks Dawid! Can you please point me to a jira which tracked the fix? > > Thanks! > Vinod > > On Thu

Re: TUMBLE function with MONTH/YEAR intervals produce milliseconds window unexpectedly

It should be fixed since version 1.6.3. Best, Dawid [1] https://issues.apache.org/jira/browse/FLINK-11017?jql=project%20%3D%20FLINK%20AND%20text%20~%20Month On Thu, 28 Mar 2019, 19:32 Vinod Mehra, wrote: > Hi All! > > We are using: org.apache.flink:flink-table_2.11:jar:1.4.2:compile > > SELEC

Support for custom triggers in Table / SQL

Hi folks, I’m trying to write a Flink job that computes a bunch of counters which requires custom triggers and I was trying to figure out the best way to express that. The query looks something like this: SELECT userId, customUDAF(...) AS counter1, customUDAF(...) AS counter2, ... FROM ( SELECT

Re: Throttling/effective back-pressure on a Kafka sink

Hi Marc, the Kafka Producer should be able to create backpressure. Could you try to increase max.block.ms to Long.MAX_VALUE? The exceptions you shared for the failure case don't look like the root causes of the problem. Could you share the full stacktraces or even full logs for this time frame. F

TUMBLE function with MONTH/YEAR intervals produce milliseconds window unexpectedly

Hi All! We are using: org.apache.flink:flink-table_2.11:jar:1.4.2:compile SELECT COALESCE(user_id, -1) AS user_id, count(id) AS count_per_window, sum(amount) AS charge_amount_per_window, TUMBLE_START(rowtime, INTERVAL '2' YEAR) AS twindow_start, TUMBLE_END(rowtime, INTERVAL '2' YEAR) AS

Re: Calcite SQL Map to Pojo Map

Based on this discussion http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/HashMap-HashSet-Serialization-Issue-td10899.html this seems by design that HashMap/Map are handled as GenericTypes . However that doesn't work with the query result table schema which generates a Map type.

Re: Calcite SQL Map to Pojo Map

Hey Rong, I don't think this is about a UDF, i reproduce the same exception with a simple map['a','b'] where the Pojo has a Map property btw for the UDF i'm already doing it (clazz is based on the specific map im creating): @Override public TypeInformation getResultType(Class[] signature) { r

Re: Calcite SQL Map to Pojo Map

If your conversion is done using a UDF you need to override the getResultType method [1] to explicitly specify the key and value type information. As generic erasure will not preseve the part of your code. Thanks, Rong [1] https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/table/udf

IllegalArgumentException when trying to execute job

Hi all, I am trying to execute a batch job that gets a list of IDs and perform a loop with a number of steps during each iteration including reading from a MS SQL Server DB. A sample pseudo-code of our implementation is the following: List ids = ... ids.foreach( id -> executeIteration();

Throttling/effective back-pressure on a Kafka sink

Hi We’ve got a job producing to a Kafka sink. The Kafka topics have a retention of 2 weeks. When doing a complete replay, it seems like Flink isn’t able to back-pressure or throttle the amount of messages going to Kafka, causing the following error: org.apache.flink.streaming.connectors.kafka.

Re: RemoteTransportException: Connection unexpectedly closed by remote task manager

I have put the task manager of the data sink log to https://gist.github.com/yinhua2018/7de42ff9c1738d5fdf9d99030db903e2 -- Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: RemoteTransportException: Connection unexpectedly closed by remote task manager

Hi Qi, I checked the JVM heap of the sink TM is low. I tried to read flink source code to identify where is exact the error happen. I think the exception happened inside DataSinkTask.invoke() // work! while (!this.taskCanceled && ((

Re: RemoteTransportException: Connection unexpectedly closed by remote task manager

Hi Yinhua, This looks like the TM executing the sink is down, maybe due to OOM or some other error. You can check the JVM heap and GC log to see if there’re any clues. Regards, Qi > On Mar 28, 2019, at 7:23 PM, yinhua.dai wrote: > > Hi, > > I write a single flink job with flink SQL with vers

Re: questions regarding offset

Thanks for answering. please see my comments below On Thu, Mar 28, 2019 at 12:32 PM Dawid Wysakowicz wrote: > Hi Avi, > > Yes, you are right. Kafka offsets are kept in state. > > Ad. 1 If you try to restore a state in a completely different > environment, and offsets are no longer compatible it

RemoteTransportException: Connection unexpectedly closed by remote task manager

Hi, I write a single flink job with flink SQL with version 1.6.1 I have one table source which read data from a database, and one table sink to output as avro format file. The table source has parallelism of 19, and table sink only has parallelism of 1. But there is always RemoteTransportExceptio

Re: What are savepoint state manipulation support plans

Thanks Gordon. We already have 5 people watching it. :-) On Thu, Mar 28, 2019 at 10:23 AM Tzu-Li (Gordon) Tai wrote: > > @Ufuk > > Yes, creating a JIRA now already to track this makes sense. > > I've proceeded to open one: https://issues.apache.org/jira/browse/FLINK-12047 > Let's move any furthe

Re: questions regarding offset

Hi Avi, Yes, you are right. Kafka offsets are kept in state. Ad. 1 If you try to restore a state in a completely different environment, and offsets are no longer compatible it will most probably fail as it won't be able to derive up to which point we already processed the records. Ad.2 What do y

Re: What are savepoint state manipulation support plans

@Ufuk Yes, creating a JIRA now already to track this makes sense. I've proceeded to open one: https://issues.apache.org/jira/browse/FLINK-12047 Let's move any further discussions there. Cheers, Gordon On Thu, Mar 28, 2019 at 5:01 PM Ufuk Celebi wrote: > I think such a tool would be really val

Re: What are savepoint state manipulation support plans

+1 On Thu, Mar 28, 2019, 5:01 AM Ufuk Celebi wrote: > I think such a tool would be really valuable to users. > > @Gordon: What do you think about creating an umbrella ticket for this > and linking it in this thread? That way, it's easier to follow this > effort. You could also link Bravo and Set

Re: RocksDBStatebackend does not write checkpoints to backup path

Hi Gordon, Thanks for your reply. I’ve found out that it should be a bug of RocksDBStateBackend [1]. [1] https://issues.apache.org/jira/browse/FLINK-12042 Best, Paul Lam > 在 2019年3月28日，17:03，Tzu-Li (Gordon) Tai 写道： > > Hi, > > Do you have

Re: RocksDBStatebackend does not write checkpoints to backup path

Hi, Do you have the full error message of the failure? A wild guess to begin with: have you made sure that there are sufficient permissions to create the directory? Best, Gordon On Tue, Mar 26, 2019 at 5:46 PM Paul Lam wrote: > Hi, > > I have a job (with Flink 1.6.4) which uses rocksdb increme

Re: What are savepoint state manipulation support plans

I think such a tool would be really valuable to users. @Gordon: What do you think about creating an umbrella ticket for this and linking it in this thread? That way, it's easier to follow this effort. You could also link Bravo and Seth's tool in the ticket as starting points. – Ufuk

Re: RocksDB local snapshot sliently disappears and cause checkpoint to fail

Ok, much clearer now. Thanks. Best Regards, Yu On Thu, 28 Mar 2019 at 15:59, Paul Lam wrote: > Hi Yu, > > I’ve set `fs.default-scheme` to hdfs, and it's mainly used for simplifying > checkpoint / savepoint / HA paths. > > And I leave the rocksdb local dir empty, so the local snapshot still goe

Re: What are savepoint state manipulation support plans

Hi! Regarding the support for savepoint reading / writing / processing directly in core Flink, we've been thinking about that lately and might push a bit for adding the functionality to Flink in the next release. For example, beside Bravo, Seth (CC'ed) also had implemented something [1] for this.

Re: What are savepoint state manipulation support plans

Hi! I dont think there is any ongoing effort in core Flink other than this library we created. You are probably right that it is pretty hacky at the moment. I would say this one way we could do it that seemed convenient to me at the time I have written the code. If you have ideas how to structur

Re: RocksDB local snapshot sliently disappears and cause checkpoint to fail

Hi Yu, I’ve set `fs.default-scheme` to hdfs, and it's mainly used for simplifying checkpoint / savepoint / HA paths. And I leave the rocksdb local dir empty, so the local snapshot still goes to YARN local cache dirs. Hope that answers your question. Best, Paul Lam > 在 2019年3月28日，15:34，Yu Li

Re: RocksDB local snapshot sliently disappears and cause checkpoint to fail

Hi Paul, Regarding "mistakenly uses the default filesystem scheme, which is specified to hdfs in the new cluster in my case", could you further clarify the configuration property and value you're using? Do you mean you're using an HDFS directory to store the local snapshot data? Thanks. Best Rega