Type Erasure - Usage of class Tuple as a type is not allowed. Use a concrete subclass (e.g. Tuple1, Tuple2, etc.
Hi,
Had asked this questions earlier as topic - "Flink - Type Erasure Exception
trying to use Tuple6 instead of Tuple"
Having issues defining a generic Tuple instead of a specific Tuple1,Tuple2
etc.
Exception in thread "main"
org.apache.flink.api.common.functions.InvalidTypesException: Usage of class
Tuple as a type is not allowed. Use a concrete subclass (e.g. Tuple1,
Tuple2, etc.) instead.
DataStream> kinesisStream = ...;
KeyedStream, Tuple> monitoringTupleKeyedStream =
kinesisStream.keyBy(new MapTupleKeySelector(groupBySet));//<= complains
about Tuple type for monitoringTupleKeyedStream
.
public static class MapTupleKeySelector implements KeySelector, Tuple> {
private final Set groupBySet;
public MapTupleKeySelector(Set groupBySet) {
this.groupBySet = groupBySet;
}
@Override
public Tuple getKey(Map inputMap) throws Exception {
int groupBySetSize = groupBySet.size();
Tuple tuple = Tuple.newInstance(groupBySetSize);
//Tuple1 tuple = new Tuple1();
int count = 0;
for (String groupBy : groupBySet) {
tuple.setField(groupByValue, count++);
}
return tuple;
}
}
Abhishek had replied back in the Thread as follows: (posting in that thread
as well creating a new thread):
However, If you are trying to build some generic framework and for
different streams, there would be different fields, you can follow the Map
approach. For the latter approach, you need to write extra mapper class
which will convert all the fields in the stream to the Map based stream.
Can I get an example of how to create this extra Mapper class ?
Currently, I am using deserialization to convert the incoming byte[] by
implementing KinesisDeserializationSchema> to convert
to a DataStream> kinesisStream.
TIA,
Re: Flink - Type Erasure Exception trying to use Tuple6 instead of Tuple
Hi,
Had asked this questions earlier as topic - "Flink - Type Erasure Exception
trying to use Tuple6 instead of Tuple"
Having issues defining a generic Tuple instead of a specific Tuple1,Tuple2
etc.
Exception in thread "main"
org.apache.flink.api.common.functions.InvalidTypesException: Usage of class
Tuple as a type is not allowed. Use a concrete subclass (e.g. Tuple1,
Tuple2, etc.) instead.
DataStream> kinesisStream = ...;
KeyedStream, Tuple> monitoringTupleKeyedStream =
kinesisStream.keyBy(new MapTupleKeySelector(groupBySet));//<= complains
about Tuple type for monitoringTupleKeyedStream
.
public static class MapTupleKeySelector implements KeySelector, Tuple> {
private final Set groupBySet;
public MapTupleKeySelector(Set groupBySet) {
this.groupBySet = groupBySet;
}
@Override
public Tuple getKey(Map inputMap) throws Exception {
int groupBySetSize = groupBySet.size();
Tuple tuple = Tuple.newInstance(groupBySetSize);
//Tuple1 tuple = new Tuple1();
int count = 0;
for (String groupBy : groupBySet) {
tuple.setField(groupByValue, count++);
}
return tuple;
}
}
Abhishek had replied back in the Thread as follows: (posting in that thread
as well creating a new thread):
However, If you are trying to build some generic framework and for
different streams, there would be different fields, you can follow the Map
approach. For the latter approach, you need to write extra mapper class
which will convert all the fields in the stream to the Map based stream.
Can I get an example of how to create this extra Mapper class ?
Currently, I am using deserialization to convert the incoming byte[] by
implementing KinesisDeserializationSchema> to convert
to a DataStream> kinesisStream.
TIA,
On Sun, Apr 7, 2019 at 5:40 PM abhishek sharma
wrote:
> I agree with Timothy, POJO would be a much better approach.
>
> However, If you are trying to build some generic framework and for
> different streams, there would be different fields, you can follow the Map
> approach. For the latter approach, you need to write extra mapper class
> which will convert all the fields in the stream to the Map based stream.
>
> Abhishek
>
> On Sun, Apr 7, 2019 at 3:07 AM Timothy Victor wrote:
>
>> Could this just be solved by creating a POJO model class for your problem?
>>
>> That is, instead of using Tuple6 - create a class that encapsulates your
>> data. This, I think, would solve your problem. But beyond that I think
>> the code will be more understandable. It's hard to have a Tuple6 of all
>> Strings, and remember what each one means -- even if I wrote the code :-)
>> Furthermore, if and when you need to add more elements to your data model,
>> you will need to refactor your entire Flink graph. Keeping a data model
>> in POJO protects against those things.
>>
>> The latter is just unsolicited code review feedback. And I know I gave
>> it without much context to your problem. So please take with a large grain
>> of salt, and if it doesn't apply just ignore it.
>>
>> Tim
>>
>>
>> On Fri, Apr 5, 2019 at 7:53 AM Chesnay Schepler
>> wrote:
>>
>>> > I tried using [ keyBy(KeySelector, TypeInformation) ]
>>>
>>> What was the result of this approach?
>>>
>>> On 03/04/2019 17:36, Vijay Balakrishnan wrote:
>>>
>>> Hi Tim,
>>> Thanks for your reply. I am not seeing an option to specify a
>>> .returns(new TypeHint>> String,String,String,String,String>>(){}) with KeyedStream ??
>>>
monitoringTupleKeyedStream = kinesisStream.keyBy(new
KeySelector() {
public Tuple getKey(Monitoring mon) throws Exception
{..return
new Tuple6<>(..}})
>>>
>>> I tried using
>>> TypeInformation>
>>> info = TypeInformation.of(new TypeHint>> String, String, String>>(){});
>>>
kinesisStream.keyBy(new KeySelector() {...}, info);
//specify typeInfo through
>>>
>>> TIA,
>>> Vijay
>>>
>>> On Tue, Apr 2, 2019 at 6:06 PM Timothy Victor wrote:
>>>
Flink needs type information for serializing and deserializing objects,
and that is lost due to Java type erasure. The only way to workaround
this is to specify the return type of the function called in the lambda.
Fabian's answer here explains it well.
https://stackoverflow.com/questions/50945509/apache-flink-return-type-of-function-could-not-be-determined-automatically-due/50947554
Tim
On Tue, Apr 2, 2019, 7:03 PM Vijay Balakrishnan
wrote:
> Hi,
> I am trying to use the KeyedStream with Tuple to handle diffrent types
> of Tuples including Tuple6.
> Keep getting the Exception:
> *Exception in thread "main"
> org.apache.flink.api.common.functions.InvalidTypesException: Usage of
> class
> Tuple as a type is not allowed. Use a concrete subclass (e.g. Tuple1,
> Tuple2, etc.) instead*.
> Is there a way around
Re: Can't build Flink for Scala 2.12
You are correct, that is a typo. Very well done for spotting it, will fix it right away. We can conclude that the current SNAPSHOT version does not build with scala 2.12 at the moment; hopefully there aren't too many issues. On 01/05/2019 08:05, Visser, M.J.H. (Martijn) wrote: In the meantime, I had a look at the Travis YAML file for examples how there the compilation for 2.12 is happening. It appears that 1) there might be a typo, because the build profile is 2.112 instead of 2.12 (see https://github.com/apache/flink/blob/master/.travis.yml#L143) and there’s also not a build profile for only 2.12, it always includes Hadoop Thanks, Martijn *From:*Visser, M.J.H. (Martijn) *Sent:* dinsdag 30 april 2019 19:09 *To:* user@flink.apache.org *Subject:* Can't build Flink for Scala 2.12 Hi all, I’m trying to build Flink (from current master branch) for Scala 2.12, using: mvn clean install -Pscala-2.12 -Dscala-2.12 -DskipTests It fails for me on the with this error: [ERROR] /home/pa35uq/Workspace/flink/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/metadata/FlinkRelMetadataQuery.scala:52: error: value EMPTY in class RelMetadataQuery cannot be accessed in object org.apache.calcite.rel.metadata.RelMetadataQuery [ERROR] Access to protected value EMPTY not permitted because [ERROR] enclosing package metadata in package plan is not a subclass of [ERROR] class RelMetadataQuery in package metadata where target is defined [ERROR] this(RelMetadataQuery.THREAD_PROVIDERS.get, RelMetadataQuery.EMPTY) [ERROR] [ERROR] Failed to execute goal net.alchim31.maven:scala-maven-plugin:3.2.2:compile (scala-compile-first) on project flink-table-planner-blink_2.12: wrap: org.apache.commons.exec.ExecuteException: Process exited with an error: 1 (Exit value: 1) -> [Help 1] Anyone an idea how to fix this? Best regards, Martijn - ATTENTION: The information in this e-mail is confidential and only meant for the intended recipient. If you are not the intended recipient, don't use or disclose it in any way. Please let the sender know and delete the message immediately. -
configuration of standalone cluster
Hi, For the first time I'm trying to set up a standalone cluster. My current configuration 4 server (1 jobmanger and 3 taskmanager) a) starting the cluster swissbib@sb-ust1:/swissbib_index/apps/flink/bin$ ./start-cluster.sh Starting cluster. Starting standalonesession daemon on host sb-ust1. Starting taskexecutor daemon on host sb-ust2. Starting taskexecutor daemon on host sb-ust3. Starting taskexecutor daemon on host sb-ust4. On the taskmanager side I get the error 2019-05-01 21:16:32,794 WARN akka.remote.ReliableDeliverySupervisor - Association with remote system [akka.ssl.tcp://flink@sb-ust1:6123] has failed, address is now gated for [50] ms. Reason: [class [B cannot be cast to class [C ([B and [C are in module java.base of loader 'bootstrap')] 2019-05-01 21:16:41,932 INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Could not resolve ResourceManager address akka.ssl.tcp://flink@sb-ust1:6123/user/resourcemanager, retrying in 1 ms: Ask timed out on [ActorSelection[Anchor(akka.ssl.tcp://flink@sb-ust1:6123/), Path(/user/resourcemanager)]] after [1 ms]. Sender[null] sent message of type "akka.actor.Identify".. 2019-05-01 21:17:01,960 INFO org.apache.flink.runtime.taskexecutor.TaskExecutor - Could not resolve ResourceManager address akka.ssl.tcp://flink@sb-ust1:6123/user/resourcemanager, retrying in 1 ms: Ask timed out on [ActorSelection[Anchor(akka.ssl.tcp://flink@sb-ust1:6123/), Path(/user/resourcemanager)]] after [1 ms]. Sender[null] sent message of type "akka.actor.Identify".. port 6123 is allowed on the jobmanager but I haven't created a specialized flink - user. - Is this necessary? if yes, is it possible to define another user for communication purposes? I followed the documentation to setup a ssl based communication (https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/security-ssl.html#example-ssl-setup-standalone-and-kubernetes) and created a keystore as described: keytool -genkeypair -alias swissbib.internal -keystore internal.keystore -dname "CN=flink.internal" -storepass verysecret -keypass verysecret -keyalg RSA -keysize 4096 and deployed the flink-conf.yaml on the whole cluster (part of flink-conf.yaml) security.ssl.internal.enabled: true security.ssl.internal.keystore: /swissbib_index/apps/flink/conf/internal.keystore security.ssl.internal.truststore: /swissbib_index/apps/flink/conf/internal.keystore security.ssl.internal.keystore-password: verysecret security.ssl.internal.truststore-password: verysecret security.ssl.internal.key-password: verysecret but this doesn't solve the problem - still no connection between task-managers and job-managers. - another question: which ports have to be enabled in the firewall for a standalone cluster? Thanks for any hints! Günter
Re: Ask about running Flink sql-client.sh
Try ./sql-client.sh embedded Rad Rad 于2019年5月1日周三 下午8:28写道： > > Hi > I would ask about the command for running sql-client.sh > > These commands don't work > ./sql-client.sh OR ./flink sql-client > > Regards. > > > > -- > Sent from: > http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/ > -- Best Regards Jeff Zhang
Ask about Running Flink Jobs From Eclipse
Hi, I can't see the running Flink job from Eclipse on Flink dashboard even I can see the running Flink jobs if I run jar file from Flink CLI or submit from Flink dashboard. Regards. -- Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
Ask about running Flink sql-client.sh
Hi I would ask about the command for running sql-client.sh These commands don't work ./sql-client.sh OR ./flink sql-client Regards. -- Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
Re: Timestamp and key preservation over operators
Hi Fabian, Guowei,
I have some updates:
1. I added timestamp extractor on all of my remaining sources (3 &
4), and the watermark does propagate to my final operator.
2. As I could not find a way to set my file sources as IDLE, I tried to
tweak the class ContinuousFileReaderOperator to be always IDLE:
/ nextElement = format.nextRecord(nextElement);
if (nextElement != null) {
readerContext.collect(nextElement);
if
(this.format.getFilePaths()[0].getPath().contains(""))
readerContext.markAsTemporarilyIdle();
} else {
/ and the result I got was there's no watermark at all for that stream, and
that IDLE status seemed not to be taken into account (my CEP operator didn't
generate any output). So I do not understand what that IDLE StreamStatus is
for.
My temporary solution, for now, is to use MAX_WATERMARK for those idle
sources. Not sure whether doing that is recommended?
Thanks for your help.
Regards,
Averell
--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
RE: Can't build Flink for Scala 2.12
In the meantime, I had a look at the Travis YAML file for examples how there the compilation for 2.12 is happening. It appears that 1) there might be a typo, because the build profile is 2.112 instead of 2.12 (see https://github.com/apache/flink/blob/master/.travis.yml#L143) and there's also not a build profile for only 2.12, it always includes Hadoop Thanks, Martijn From: Visser, M.J.H. (Martijn) Sent: dinsdag 30 april 2019 19:09 To: user@flink.apache.org Subject: Can't build Flink for Scala 2.12 Hi all, I'm trying to build Flink (from current master branch) for Scala 2.12, using: mvn clean install -Pscala-2.12 -Dscala-2.12 -DskipTests It fails for me on the with this error: [ERROR] /home/pa35uq/Workspace/flink/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/metadata/FlinkRelMetadataQuery.scala:52: error: value EMPTY in class RelMetadataQuery cannot be accessed in object org.apache.calcite.rel.metadata.RelMetadataQuery [ERROR] Access to protected value EMPTY not permitted because [ERROR] enclosing package metadata in package plan is not a subclass of [ERROR] class RelMetadataQuery in package metadata where target is defined [ERROR] this(RelMetadataQuery.THREAD_PROVIDERS.get, RelMetadataQuery.EMPTY) [ERROR] [ERROR] Failed to execute goal net.alchim31.maven:scala-maven-plugin:3.2.2:compile (scala-compile-first) on project flink-table-planner-blink_2.12: wrap: org.apache.commons.exec.ExecuteException: Process exited with an error: 1 (Exit value: 1) -> [Help 1] Anyone an idea how to fix this? Best regards, Martijn - ATTENTION: The information in this e-mail is confidential and only meant for the intended recipient. If you are not the intended recipient, don't use or disclose it in any way. Please let the sender know and delete the message immediately. -
