import org.datavec.api.split.InputSplit; //导入依赖的package包/类
/**
 * Initialize all required jdbc elements and make the reader ready for iteration.
 *
 * Possible configuration keys :
 * <ol>
 *     <li>JDBCRecordReader.TRIM_STRINGS : Whether or not read strings should be trimmed before being returned. False by default</li>
 *     <li>JDBCRecordReader.JDBC_URL : Jdbc url to use for datastource configuration (see JDBCRecordReaderTest for examples)</li>
 *     <li>JDBCRecordReader.JDBC_DRIVER_CLASS_NAME : Driver class to use for datasource configuration</li>
 *     <li>JDBCRecordReader.JDBC_USERNAME && JDBC_PASSWORD : Username and password to use for datasource configuration</li>
 *     <li>JDBCRecordReader.JDBC_RESULTSET_TYPE : ResultSet type to use (int value defined in jdbc doc)</li>
 * </ol>
 *
 * Url and driver class name are not mandatory. If one of them is specified, the other must be specified as well. If
 * they are set and there already is a DataSource set in the reader, it will be discarded and replaced with the
 * newly created one.
 *
 * @param conf a configuration for initialization
 * @param split not handled yet, will be discarded
 */
@Override
public void initialize(Configuration conf, InputSplit split) throws IOException, InterruptedException {
    this.setConf(conf);
    this.setTrimStrings(conf.getBoolean(TRIM_STRINGS, trimStrings));
    this.setResultSetType(conf.getInt(JDBC_RESULTSET_TYPE, resultSetType));

    String jdbcUrl = conf.get(JDBC_URL);
    String driverClassName = conf.get(JDBC_DRIVER_CLASS_NAME);
    // url and driver must be both unset or both present
    if (jdbcUrl == null ^ driverClassName == null) {
        throw new IllegalArgumentException(
            "Both jdbc url and driver class name must be provided in order to configure JDBCRecordReader's datasource");
    }
    // Both set, initialiaze the datasource
    else if (jdbcUrl != null) {
        // FIXME : find a way to read wildcard properties from conf in order to fill the third argument bellow
        this.dataSource = new DriverDataSource(jdbcUrl, driverClassName, new Properties(), conf.get(JDBC_USERNAME),
            conf.get(JDBC_PASSWORD));
    }
    this.initializeJdbc();
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
@Override
public void initialize(Configuration conf, InputSplit split) throws IOException, InterruptedException {
    this.appendLabel = conf.getBoolean(APPEND_LABEL, false);
    this.labels = new ArrayList<>(conf.getStringCollection(LABELS));
    this.height = conf.getInt(HEIGHT, height);
    this.width = conf.getInt(WIDTH, width);
    this.channels = conf.getInt(CHANNELS, channels);
    this.cropImage = conf.getBoolean(CROP_IMAGE, cropImage);
    if ("imageio".equals(conf.get(IMAGE_LOADER))) {
        this.imageLoader = new ImageLoader(height, width, channels, cropImage);
    } else {
        this.imageLoader = new NativeImageLoader(height, width, channels, imageTransform);
    }
    this.conf = conf;
    initialize(split);
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
protected void doInitialize(InputSplit split) {

        if (labels == null && appendLabel) {
            URI[] locations = split.locations();
            if (locations.length > 0) {
                //root dir relative to example where the label is the parent directory and the root directory is
                //recursively the parent of that
                File parent = new File(locations[0]).getParentFile().getParentFile();
                //calculate the labels relative to the parent file
                labels = new ArrayList<>();

                for (File labelDir : parent.listFiles())
                    labels.add(labelDir.getName());
            }
        }
        locationsIterator = split.locationsPathIterator();
    }
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
@Test
public void testReadingJson() throws Exception {
    //Load 3 values from 3 JSON files
    //stricture: a:value, b:value, c:x:value, c:y:value
    //And we want to load only a:value, b:value and c:x:value
    //For first JSON file: all values are present
    //For second JSON file: b:value is missing
    //For third JSON file: c:x:value is missing

    ClassPathResource cpr = new ClassPathResource("json/json_test_0.txt");
    String path = cpr.getFile().getAbsolutePath().replace("0", "%d");

    InputSplit is = new NumberedFileInputSplit(path, 0, 2);

    RecordReader rr = new JacksonRecordReader(getFieldSelection(), new ObjectMapper(new JsonFactory()));
    rr.initialize(is);

    testJacksonRecordReader(rr);
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
public void initialize(InputSplit split) throws IOException, InterruptedException {
    if(split instanceof ListDoubleSplit) {
        ListDoubleSplit listDoubleSplit = (ListDoubleSplit)split;
        this.delimitedData = listDoubleSplit.getData();
        this.dataIter = this.delimitedData.iterator();
    } else {
        throw new IllegalArgumentException("Illegal type of input split " + split.getClass().getName());
    }
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
@Override
public DataSetIterator create(InputSplit inputSplit) {
    try {
        return createInternal(inputSplit);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
private DataSetIterator createInternal(InputSplit inputSplit) throws IOException {
    ImageTransform imageTransform = imageTransformFactory.create();
    int width = imageTransformConfigurationResource.getScaledWidth();
    int height = imageTransformConfigurationResource.getScaledHeight();
    int channels = imageTransformConfigurationResource.getChannels();
    int batchSize = networkConfigurationResource.getBatchSize();
    int outputs = networkConfigurationResource.getOutputs();
    ImageRecordReader recordReader = new ImageRecordReader(height, width, channels, pathLabelGenerator);
    recordReader.initialize(inputSplit, imageTransform);
    RecordReaderDataSetIterator recordReaderDataSetIterator = new RecordReaderDataSetIterator(recordReader, batchSize, 1, outputs);
    DataNormalization scaler = new ImagePreProcessingScaler(0, 1);
    scaler.fit(recordReaderDataSetIterator);
    recordReaderDataSetIterator.setPreProcessor(scaler);
    return recordReaderDataSetIterator;
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
@Override
public InputSplitResource getInputSplit() {
    log.info("Select images location");
    File file = storeChooser.getDirectory()
            .orElseThrow(() -> new ImageStoreException("No store chosen"));
    FileSplit fileSplit = new FileSplit(file, NativeImageLoader.ALLOWED_FORMATS, random);
    BalancedPathFilter pathFilter = new BalancedPathFilter(random, NativeImageLoader.ALLOWED_FORMATS, pathLabelGenerator);

    InputSplit[] inputSplit = fileSplit.sample(pathFilter, trainWeight, testWeight);

    InputSplitResource resource = new InputSplitResource();
    resource.setTrain(inputSplit[0]);
    resource.setTest(inputSplit[1]);
    return resource;
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
@Override
public void initialize(Configuration conf, InputSplit split) throws IOException, InterruptedException {
    this.conf = conf;
    this.appendLabel = conf.getBoolean(APPEND_LABEL, false);
    this.height = conf.getInt(HEIGHT, height);
    this.width = conf.getInt(WIDTH, width);
    if ("imageio".equals(conf.get(IMAGE_LOADER))) {
        this.imageLoader = new ImageLoader(height, width);
    } else {
        this.imageLoader = new NativeImageLoader(height, width);
    }

    initialize(split);
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
public RecordReader getRecordReader(int batchSize, int numExamples, int[] imgDim, int numLabels,
                PathLabelGenerator labelGenerator, boolean train, double splitTrainTest, Random rng) {
    load(batchSize, numExamples, numLabels, labelGenerator, splitTrainTest, rng);
    RecordReader recordReader =
                    new ImageRecordReader(imgDim[0], imgDim[1], imgDim[2], labelGenerator, imageTransform);

    try {
        InputSplit data = train ? inputSplit[0] : inputSplit[1];
        recordReader.initialize(data);
    } catch (IOException | InterruptedException e) {
        e.printStackTrace();
    }
    return recordReader;
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
@Test
public void testLfwReader() throws Exception {
    String subDir = "lfw-a/lfw";
    File path = new File(FilenameUtils.concat(System.getProperty("user.home"), subDir));
    FileSplit fileSplit = new FileSplit(path, LFWLoader.ALLOWED_FORMATS, new Random(42));
    BalancedPathFilter pathFilter = new BalancedPathFilter(new Random(42), LFWLoader.LABEL_PATTERN, 1, 1, 1);
    InputSplit[] inputSplit = fileSplit.sample(pathFilter, 1);
    RecordReader rr = new ImageRecordReader(250, 250, 3, LFWLoader.LABEL_PATTERN);
    rr.initialize(inputSplit[0]);
    List<String> exptedLabel = rr.getLabels();

    RecordReader rr2 = new LFWLoader(new int[] {250, 250, 3}, true).getRecordReader(1, 1, 1, new Random(42));

    assertEquals(exptedLabel.get(0), rr2.getLabels().get(0));
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
@Override
public void initialize(Configuration conf, InputSplit split) throws IOException, InterruptedException {
    this.conf = conf;
    this.appendLabel = conf.getBoolean(APPEND_LABEL, false);
    this.labels = new ArrayList<>(conf.getStringCollection(LABELS));
    initialize(split);
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
@Override
protected int poll() throws Exception {
    Exchange exchange = endpoint.createExchange();
    InputSplit split = inputFromExchange(exchange);
    RecordReader reader = inputFormat.createReader(split, configuration);
    int numMessagesPolled = 0;
    while (reader.hasNext()) {
        // create a message body
        while (reader.hasNext()) {
            exchange.getIn().setBody(reader.next());

            try {
                // send message to next processor in the route
                getProcessor().process(exchange);
                numMessagesPolled++; // number of messages polled
            } finally {
                // log exception if an exception occurred and was not handled
                if (exchange.getException() != null) {
                    getExceptionHandler().handleException("Error processing exchange", exchange,
                                    exchange.getException());
                }
            }
        }


    }

    return numMessagesPolled;
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
/**
 * @param exchange
 * @return
 */
@Override
public InputSplit getSplit(Exchange exchange) {
    List<List<String>> data = (List<List<String>>) exchange.getIn().getBody();
    InputSplit listSplit = new ListStringSplit(data);
    return listSplit;
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
/**
 * Creates a reader from an input split
 *
 * @param split the split to read
 * @return the reader from the given input split
 */
@Override
public RecordReader createReader(InputSplit split) throws IOException, InterruptedException {
    RecordReader reader = new ListStringRecordReader();
    reader.initialize(split);
    return reader;
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
@Override
public RecordReader createReader(InputSplit split) throws IOException, InterruptedException {
    LineRecordReader ret = new LineRecordReader();
    ret.initialize(split);
    return ret;

}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
@Override
public RecordReader createReader(InputSplit split) throws IOException, InterruptedException {
    CSVRecordReader ret = new CSVRecordReader();
    ret.initialize(split);
    return ret;

}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
@Override
public void initialize(Configuration conf, InputSplit split) throws IOException, InterruptedException {
    appendLabel = conf.getBoolean(APPEND_LABEL, true);
    doInitialize(split);
    this.inputSplit = split;
    this.conf = conf;
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
/**
 * Called once at initialization.
 *
 * @param split the split that defines the range of records to read
 * @throws IOException
 * @throws InterruptedException
 */
@Override
public void initialize(InputSplit split) throws IOException, InterruptedException {
    if (split instanceof ListStringSplit) {
        ListStringSplit listStringSplit = (ListStringSplit) split;
        delimitedData = listStringSplit.getData();
        dataIter = delimitedData.iterator();
    } else {
        throw new IllegalArgumentException("Illegal type of input split " + split.getClass().getName());
    }
}
 

import org.datavec.api.split.InputSplit; //导入依赖的package包/类
@Override
public void initialize(InputSplit split) throws IOException, InterruptedException {
    if (split instanceof FileSplit)
        throw new UnsupportedOperationException("Cannot use JacksonRecordReader with FileSplit");
    this.uris = split.locations();
    if (shuffle) {
        List<URI> list = Arrays.asList(uris);
        Collections.shuffle(list, r);
        uris = list.toArray(new URI[uris.length]);
    }
}