apache-flink - 在 Apache Flink 中从 HDFS 地址流式传输文件

Question

在我的 Flink 代码中，我正在流式传输位于 HDFS 文件夹上的文件，我收到错误“(没有此类文件或目录)”，但是我确信文件名和地址是正确的，因为我使用了相同的文件名和地址在批处理方法中，一切都很顺利。 有谁知道可能是什么问题？ 这是我的代码:

DataStream<FebrlObject> myStream = 
env.addSource(new MyObjectGenerator("hdfs://../Data/Dataset1.csv"));

及其相关类:

public class MyObjectGenerator implements SourceFunction<MyObject> {

    private String dataFilePath;
    private float servingSpeedFactor;
    private Integer rowNo ; 
    private transient BufferedReader reader;
    private transient InputStream inputStream;

    public MyObjectGenerator(String dataFilePath) {
        this(dataFilePath, 1.0f);
    }

    public MyObjectGenerator(String dataFilePath, float servingSpeedFactor) {
        this.dataFilePath = dataFilePath;
        this.servingSpeedFactor = servingSpeedFactor;
        rowNo = 0 ;
    }

    @Override
    public void run(SourceContext<MyObject> sourceContext) throws Exception {
        long servingStartTime = Calendar.getInstance().getTimeInMillis();
        inputStream = new DataInputStream(new FileInputStream(dataFilePath));
        reader = new BufferedReader(new InputStreamReader(inputStream));
        String line;
        long dataStartTime;
        rowNo++;
        if (reader.ready() && (line = reader.readLine()) != null ) {
            MyObject myObject = MyObject.fromString(line);
            if (febrlObject!= null )
            sourceContext.collect(myObject);
        } else {
            return;
        }
        while (reader.ready() && (line = reader.readLine()) != null) {
            MyObject myObject = MyObject.fromString(line);
            sourceContext.collect( febrlObject );
        }
        this.reader.close();
        this.reader = null;
        this.inputStream.close();
        this.inputStream = null;
    }

    @Override
    public void cancel() {
        try {
            if (this.reader != null) {
                this.reader.close();
            }
            if( this.inputStream != null) {
                this.inputStream.close();
            }
        } catch (IOException ioe) {
            //
        } finally {
            this.reader = null;
            this.inputStream = null;
        }
    }
}

Answer 1

您尝试使用 Java 的常规 FileInputStream 访问 HDFS 中的文件。 FileInputStream can only access the local file system. It does not know anything about talking to HDFS. You need to use the HDFS client to read files from HDFS. See Flink's以 FileInputFormat` 为例。

但是，如果可能的话，我会尽量避免自己实现这一点。您可以尝试使用Flink的FileInputFormat逐行读取文件(返回 DataStream<String> )和解析该行的连续(平面)映射器。