在Apache Flink中创建数据流时出错

Question

使用fromElements函数创建数据流时出错

下面是探险-

原因：java.io.IOException：无法从源反序列化元素。如果您使用的是用户定义的序列化（值和可写类型），请检查序列化函数。序列化程序是org.apache.flink.api.java.typeutils.runtime.kryo.kryoSerializer@599fcdda在org.apache.flink.streaming.api.functions.source.fromelementsfunction.run(fromelementsfunction.java:121)在org.apache.flink.streaming.api.operators.streamsource.run(streamsource.java:58)在

Answer 1

为什么要处理InputStreamReader元组？我猜这里有些理解不到位。泛型类型指定要处理的数据的类型。例如：

DataStream<Integer> intStream = env.fromElements(1, 2, 3, 4, 5);

生成包含5个integer元组的有限数据流。我假设您实际上希望使用InputStreamReader来生成实际的元组。

如果要通过HTTPURLConnection进行读取，可以实现自己的SourceFunction（或RichSourceFunction)，如下所示（将Out替换为要使用的实际数据类型--还可以考虑FlinksTuple0到Tuple25类型）：

env.addSource(new SourceFunction<OUT> {
    private volatile boolean isRunning = true;

    @Override
    public void run(SourceContext<OUT> ctx) {
        InputStreamReader isr = null;
        try {
            URL url = new URL("ex.in/res");
            HttpURLConnection httpconn = (HttpURLConnection) url.openConnection();
            if (httpconn.getResponseCode() != 200)
                throw new RuntimeException("Failed : HTTP error code : " + httpconn.getResponseCode());
            isr = new InputStreamReader((httpconn.getInputStream()));
        } catch (Exception e) {
            // clean up; log error
            return;
        }

        while(isRunning) {
            OUT tuple = ... // get data from isr
            ctx.collect(tuple);
        }
    }

    @Override
    public void cancel() {
         this.isRunning = false;
    }
});