@Override
    public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo,
                                 StreamExecutionEnvironment env,
                                 StreamTableEnvironment tableEnv) {
        FileSource fileSource = new FileSource();
        FileSourceTableInfo tableInfo = (FileSourceTableInfo) sourceTableInfo;

        fileSource.initSource(tableInfo);

        String fields = StringUtils.join(tableInfo.getFields(), ",");

        DataStreamSource<Row> source = env.addSource(fileSource, tableInfo.getOperatorName(), tableInfo.getTypeInformation());

        return tableEnv.fromDataStream(source, fields);
    }
rps Record Per Second: deserialize data and out record num
file/file-source/src/main/java/com/dtstack/flink/sql/source/file/FileSource.java