FileSystemInput implements BatchInput, ProvidesAlias, ProvidesValidations,
    InstantiatesComponents, UsesExpectedSchema, DeclaresProvidingSchema, CanReturnErroredData { private Dataset<Row> readInputFormat(String path) throws Exception { LOG.debug("Reading InputFormat[{}]: {}", inputType, path); Class<? extends InputFormat> inputFormatClass = Class.forName(inputType).asSubclass(InputFormat.class); TranslateFunction translateFunction = getTranslateFunction(translatorConfig); Dataset<Row> encoded = getEncodedRowsFromInputFormat(path, inputFormatClass); TranslationResults results = new TranslationResults( encoded.javaRDD().flatMap(translateFunction), translateFunction.getProvidingSchema(), getProvidingSchema()); errors = results.getErrored(); return results.getTranslated(); }  @Override String getAlias(); @Override void configure(Config config); @Override Dataset<Row> read(); @Override void receiveExpectedSchema(StructType expectedSchema); @Override StructType getProvidingSchema(); @Override Validations getValidations(); @Override Set<InstantiatedComponent> getComponents(Config config, boolean configure); @Override Dataset<Row> getErroredData(); static final String FORMAT_CONFIG; static final String PATH_CONFIG; static final String CSV_HEADER_CONFIG; static final String CSV_SEPARATOR_CONFIG; static final String CSV_ENCODING_CONFIG; static final String CSV_QUOTE_CONFIG; static final String CSV_ESCAPE_CONFIG; static final String CSV_COMMENT_CONFIG; static final String CSV_INFER_SCHEMA_CONFIG; static final String CSV_IGNORE_LEADING_CONFIG; static final String CSV_IGNORE_TRAILING_CONFIG; static final String CSV_NULL_VALUE_CONFIG; static final String CSV_NAN_VALUE_CONFIG; static final String CSV_POS_INF_CONFIG; static final String CSV_NEG_INF_CONFIG; static final String CSV_DATE_CONFIG; static final String CSV_TIMESTAMP_CONFIG; static final String CSV_MAX_COLUMNS_CONFIG; static final String CSV_MAX_CHARS_COLUMN_CONFIG; static final String CSV_MAX_MALFORMED_LOG_CONFIG; static final String CSV_MODE_CONFIG; static final String INPUT_FORMAT_TYPE_CONFIG; static final String CSV_FORMAT; static final String PARQUET_FORMAT; static final String JSON_FORMAT; static final String INPUT_FORMAT_FORMAT; static final String TEXT_FORMAT; static final String TRANSLATOR_CONFIG; static final String SCHEMA_CONFIG; }
@Test public void readInputFormat() throws Exception { Map<String, Object> paramMap = new HashMap<>(); paramMap.put(FileSystemInput.FORMAT_CONFIG, "input-format"); paramMap.put(FileSystemInput.PATH_CONFIG, FileSystemInput.class.getResource(CSV_DATA).getPath()); paramMap.put(FileSystemInput.INPUT_FORMAT_TYPE_CONFIG, TextInputFormat.class.getCanonicalName()); paramMap.put("translator" + "." + ComponentFactory.TYPE_CONFIG_NAME, DummyInputFormatTranslator.class.getCanonicalName()); config = ConfigFactory.parseMap(paramMap); FileSystemInput formatInput = new FileSystemInput(); assertNoValidationFailures(formatInput, config); formatInput.configure(config); Dataset<Row> results = formatInput.read(); assertEquals("Invalid number of rows", 4, results.count()); assertEquals("Invalid first row result", 0L, results.first().getLong(0)); assertEquals("Invalid first row result", "One,Two,Three,Four", results.first().getString(1)); }