Pyflink implements custom sourcefunction

brief introduction

pyflink At present, it is impossible to be like Map、FlatMap Same definition python UDF To achieve Source UDF Of , Instead, you need to implement Java SourceFunction, And then in python Introduction in operation .

// pyflink in SourceFunction The definition of class SourceFunction(JavaFunctionWrapper):    """    Base class for all stream data source in Flink.    """    def __init__(self, source_func: Union[str, JavaObject]):        """        Constructor of SinkFunction.        :param source_func: The java SourceFunction object.        """        super(SourceFunction, self).__init__(source_func)

Method 1

First of all use first java Simple implementation SourceFunction.

public class MyCustomSourceFunction implements SourceFunction<Row> {    private static final String[] NAMES = {"Bob", "Marry", "Henry", "Mike", "Ted", "Jack"};    public void run(SourceContext sourceContext) {        Random random = new Random();        for (int i = 0; i < NAMES.length; i++) {            Row row = Row.of(i, NAMES[i]);            sourceContext.collect(row);        }    }    @Override    public void cancel() {}}

After that python Introduction in operation SourceFunction

#  Fill in the classpath directly , Realize to java source References to custom_source = SourceFunction("org.apache.flink.python.util.MyCustomSourceFunction") # add_sourceds = env.add_source(custom_source, type_info=Types.ROW([Types.INT(), Types.STRING()]))

Method 2

Method 1 has been able to implement simple Source 了 , But there is a problem , Method 1 cannot pass parameters , For example, to achieve a RedisSource, Hope to pass python Job transfer parameters , Such as redis Of host and port, At this point, method 1 cannot be achieved , So we have method two .

package com.test;public class MyRedisSourceFunction implements SourceFunction<Row> {    private string host;    private int port;    public MyRedisSourceFunction(string host, int port) {        this.host = host;        this.port = port;    }        public void run(SourceContext sourceContext) {         Jedis jedis = xxxx; //  initialization jedis         while(true) {              //  Connect redis, get data          }    }    @Override    public void cancel() {}}

here python How to write ？ First you need to define a Python class, Initialize internally Java Redis Source.

#  Realization Python sourcefunction, Internal initialization java sourcefunctionclass PyRedisSourceFunction(SourceFunction):    def __init__(self, host, port):        #  obtain Java RedisSource class         JMyRedisSourceFunction = get_gateway().jvm.com.test.MyRedisSourceFunction                #  initialization Java redis source        j_redis_source = JMyRedisSourceFunction(host, port)                #  Call the parent class __init__        super(PyRedisSourceFunction, self).__init__(sink_func=j_redis_source)

And then in main Function Python class

host = 127.0.0.1port = 12345#  initialization Python classredis_source = PyRedisSourceFunction(host, port) # add_sourceds = env.add_source(redis_source, type_info=Types.STRING())