where's socketTextStream in pyflink

Admin 1
Viewed 291

I want to translate the following code into pyflink and run it in pyflink-shell.sh afterwards.

public class MapDemo {
    private static int index = 1;
    public static void main(String[] args) throws Exception {
        //1.获取执行环境配置信息
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.定义加载或创建数据源（source）,监听9000端口的socket消息
        DataStream<String> textStream = env.socketTextStream("localhost", 9000, "\n");
        //3.map操作。
        DataStream<String> result = textStream.map(s -> (index++) + ".您输入的是：" + s);
        //4.打印输出sink
        result.print();
        //5.开始执行
        env.execute();
    }

But I can not found socketTextStreamin b_env,bt_env,s_env,st_env

so where's the socketTextStream in pyflink api?

pyflink apache-flink
2 Answers
user_20008230

Yes, you can use socketTextStream in pyflink, though it has not been officially supported yet. Example:

from pyflink.datastream import DataStream, StreamExecutionEnvironment

if __name__ == '__main__':
    s_env = StreamExecutionEnvironment.get_execution_environment()
    socket_stream = DataStream(s_env._j_stream_execution_environment.socketTextStream('localhost', 9999))
    socket_stream.print()
    s_env.execute('socket_stream')
user_70266850
Related