why pass arguments to the constructor of operator function class is null of Flink?

Question

I am studying Flink, I want to build an operator function which extends ProcessWindowFunction and overload a new constructor with a parameter as a field value of the class, but when this class is instanced, without of this field, I am confused. code as follow.

import com.aliyun.datahub.client.model.Field;
import com.aliyun.datahub.client.model.FieldType;
import com.aliyun.datahub.client.model.PutRecordsResult;
import io.github.streamingwithflink.chapter8.PoJoElecMeterSource;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class DataHubSinkDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        env.enableCheckpointing(10_000L);
        env.setParallelism(2);

        RecordSchemaSer schema = new RecordSchemaSer();

        schema.addField(new Field("id", FieldType.STRING));            

        DataStream<PutRecordsResult> out = env
                .addSource(new PoJoElecMeterSource())
                .keyBy( r -> r.getId())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(3))) 
                .process(new PutDatahubFunction<>(schema));  // PutDatahubFunction is my build a new Operator function class

        env.execute();
    }
}

variable schema is a parameter which I want to send to the constructor, it is an instance of RecordSchemaSer Class

import com.aliyun.datahub.client.model.RecordSchema;
import java.io.Serializable;

public class RecordSchemaSer
        extends RecordSchema
        implements Serializable {

}

PutDatahubFunction is a class extends ProcessWindowFunction, code as follows

import com.aliyun.datahub.client.model.*;

import io.github.streamingwithflink.chapter8.PUDAPoJo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;

public class PutDatahubFunction<IN extends PUDAPoJo, KEY>
        extends ProcessWindowFunction<IN, PutRecordsResult, KEY, TimeWindow> {

    private DataHubBase dataHubHandler;
    private List<RecordEntry> recordEntries;
    private RecordSchema schema;

    public PutDatahubFunction(RecordSchema schema) {

        this.schema = schema;
        System.out.println("field 'id' not exist ? " + this.schema.containsField("id"));  // it's true
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        .........
    }

    @Override
    public void process(KEY KEY,
                        Context context,
                        Iterable<IN> elements,
                        Collector<PutRecordsResult> out)
            throws Exception {

        RecordEntry entry = new RecordEntry();

        for (IN e : elements) {
            System.out.println("field 'id' not exist ? " + this.schema.containsField("id")); // it's false
            ......
        }

    }
}

the first system.out in the constructor, this.schema.containsField("id") is true, but the second system.out in process method, this.schema.containsField("id") is false! why? I have system.out two class name of the instance which both are PutDatahubFunction.

use ValueState not working, because constructor not call getRuntimeContext(), otherwise Exception in thread "main" java.lang.IllegalStateException: The runtime context has not been initialized. code as follow:

private ValueState<RecordSchema>  schema;


public PutTupleDatahubFunction(RecordSchema schema) throws IOException {
    ValueStateDescriptor schemaDes =
            new ValueStateDescriptor("datahub schema", TypeInformation.of(RecordSchema.class));
    /*
     * error Exception in thread "main" java.lang.IllegalStateException:
     * The runtime context has not been initialized.
     */
    this.schema = getRuntimeContext().getState(schemaDes);
    this.schema.update(schema);
}

I am very fuzzing, who can tell me the reason, Is there any way to pass arguments to the constructor of this operator function class? thanks.

I think this problem could be solved with ValueState, but I didn't think use ValueState because State consumes JVM, and I want to find out the reason. — kursk.ye

kursk.ye kursk.ye · Accepted Answer · 2020-05-13T03:09:14

I finally figured out why，the reason is Serialize and Deserialize. I am not coding RecordSchemaSer reason is Serialize content, due to null

public class RecordSchemaSer
        extends RecordSchema
        implements Serializable
{


}

why pass arguments to the constructor of operator function class is null of Flink?

1 Answers