Hello Fabian,

Thanks for this quick answer ! I understand I cannot mix the custom map 
function with the lambda definition. The workaround we found for the issue was 
to pass the function to get the key as a parameter to the CustomMapFunction 
instead of the key itself, and now it works !

Regards,

Isabelle

De : Fabian Hueske [mailto:fhue...@gmail.com]
Envoyé : jeudi 31 mai 2018 09:34
À : Robin, Isabelle <isabelle.ro...@ercom.fr>
Cc : user@flink.apache.org; Nejjar, Driss <driss.nej...@ext.bleckwen.ai>
Objet : Re: Use element of the DataStream in parameter of RichMapFunction (open 
function not called)

Hi Isabelle,
Welcome to the Flink user mailing list!
You are mixing up the two ways to specify a function:
1. Defining a function as a class / object and passing an instance in the map() 
method. Given your CustomMapFunction class, this looks as follows:
stream.keyBy(...).map(new CustomMapFunction())
2. Defining a function as a lambda function. In this case, you don't have to 
define an extra class (CustomMapFunction in your case) but rather give the 
logic that you would put into the MapFunction.map() method as a lambda function.
Internally, the lambda function is "wrapped" in a MapFunction. This style looks 
as follows:
stream.keyBy(...).map(t => [directly apply your map() logic and return a 
Transaction])

In your code, you define a lambda function (by using "transaction => ...") that 
returns a CustomMapFunction, instead of a Transaction. Hence, the first program 
fails due to an invalid type.
I would recommend to use the 1. class/object style for complex logic and the 2. 
lambda function style for simple transformations, filters, key-extractions, ...
Btw. your key extractor (transaction => Feature.transaction).getKey) is not 
very efficient because it creates a new Feature object, for each transaction.
Best, Fabian


2018-05-30 17:38 GMT+02:00 Robin, Isabelle 
<isabelle.ro...@ercom.fr<mailto:isabelle.ro...@ercom.fr>>:
Hello,

I’m working with Flink 1.4.2 (Scala API) and I’m having some trouble with my 
custom RichMapFunction as I want the element in my Datastream to also be used 
for a parameter of this custom class. My RichMapFunction is a simple counter 
based on a MapState

Let’s say I have those classes

-          case class Feature(transaction: Transaction) { override def getKey: 
(String, String) = … }

-          class CustomMapFunction(featureKey: (String, String)) extends 
RichMapFunction[Transaction, Transaction]

I implemented my custom map function with the needed functions but I 
encountered different issues as I tried several solutions for this. In the 
following chunks of code, stream is a DataStream[Transaction] and I expect a 
DataStream[Transaction] as output type too


•         stream.keyBy(transaction => 
Feature(transaction).getKey).map(transaction => new 
CustomMapFunction(Feature(transaction).getKey))

o   this leads to a compilation error (“Expression of type CustomMapFunction 
doesn’t conform to expected type R_”), which, as far as I understand, should 
come from the fact I’m already using transaction for the Feature(transaction) 
part

•         stream.keyBy(transaction => 
Feature(transaction).getKey).map(transaction => new 
CustomMapFunction(Feature(transaction).getKey).map(transaction))

o   compiles but fails with a NullPointerException at runtime as the MapState 
is not initialized. When running with debugger the open function was not used 
which leads the MapState to stay null (I don’t have this problem with a more 
simple version of my CustomMapFunction which does not need this parameter based 
on the transaction)

Do you have an idea of how I could solve this issue ?

Thanks in advance for any help and I hope I was clear enough (that’s my first 
question on the mailing list, don’t hesitate to say if I forgot some steps or 
elements ☺)

Best regards,

Isabelle

Reply via email to