Re: Change to StreamingFileSink in Flink 1.10

Thanks @Seth Wiesman and all. -- Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Change to StreamingFileSink in Flink 1.10

Thanks @Seth Wiesman Ah，I just found I used 1.10-SNAPSHOT locally so can not reproduce the bug. @Averell you can use casts first and wait for 1.10.1 version, 1.10.1 will release soon. Best, Leonard > 在 2020年4月21日，22:03，Seth Wiesman 写道： > > Hi All, > > There is a bug in the builder that prev

Re: Change to StreamingFileSink in Flink 1.10

Hi All, There is a bug in the builder that prevents it from compiling in scala due to differences in type inference between java and scala[1]. It as already been resolved for 1.10.1 and 1.11. In the meantime, just go ahead and use casts or construct the object in a java class. Seth [1] https://i

Re: Change to StreamingFileSink in Flink 1.10

Hi, Averell I found you’re using scala so I reproduced your case local in Scala 2.11.12 with Flink 1.10.0 and it works too. From your picture it’s wired that line`.withBucketAssigner(new DateTimeBucketAssigner)` hint is `Any`, it should be `RowFormatBuilder` otherwise you can not call `#build

Re: Change to StreamingFileSink in Flink 1.10

Hello Leonard, Sivaprasanna, But my code was working fine with Flink v1.8. I also tried with a simple String DataStream, and got the same error. /StreamingFileSink .forRowFormat(new Path(path), new SimpleStringEncoder[String]()) .withRollingPolicy(DefaultRollingPolicy.b

Re: Change to StreamingFileSink in Flink 1.10

I agree with Leonard. I have just tried the same in Scala 2.11 with Flink 1.10.0 and it works just fine. Cheers, Sivaprasanna On Tue, Apr 21, 2020 at 12:53 PM Leonard Xu wrote: > Hi, Averell > > I guess it’s none of `#withRollingPolicy` and `#withBucketAssigner` and > may cause by generics typ

Re: Change to StreamingFileSink in Flink 1.10

Hi, Averell I guess it’s none of `#withRollingPolicy` and `#withBucketAssigner` and may cause by generics type that your Encoder’s element type(IN) does not match BucketAssigner element type(IN) or you lost the generics type information when instantiate them. Could you post more code phase?

Re: Change to StreamingFileSink in Flink 1.10

Hi, I tried to add the following cast, and it works. Doesn't look nice though /StreamingFileSink .forRowFormat(new Path(path), myEncoder) .withRollingPolicy(DefaultRollingPolicy.create().build()) .withBucketAssigner(myBucketAssigner)*.asInstan

Re: Change to StreamingFileSink in Flink 1.10

Hi Sivaprasanna, That is a compile-time error, not a runtime error. /value build is not a member of ?0 possible cause: maybe a semicolon is missing before `value build'?/. There won't be any issue with either *withRollingPolicy*() or /withBucketAssigner/(), but not both. Thanks and regards, Av

Re: Change to StreamingFileSink in Flink 1.10

Hi Averell, Can you please the complete stacktrace of the error? On Mon, Apr 20, 2020 at 4:48 PM Averell wrote: > Hi, > > I have the following code: > / StreamingFileSink > .forRowFormat(new Path(path), myEncoder) > .withRollingPolicy(DefaultRollingPolicy.

Change to StreamingFileSink in Flink 1.10

Hi, I have the following code: / StreamingFileSink .forRowFormat(new Path(path), myEncoder) .withRollingPolicy(DefaultRollingPolicy.create().build()) .withBucketAssigner(myBucketAssigner) .build()/ This is working fine in Flink 1.8