@Test public void testInputFormat() throws Exception { final RyaStatement input = RyaStatement.builder() .setSubject(new RyaIRI("http: .setPredicate(new RyaIRI("http: .setObject(new RyaIRI("http: .setColumnVisibility(new byte[0]) .setValue(new byte[0]) .build(); apiImpl.add(input); final Job jobConf = Job.getInstance(); GraphXInputFormat.setMockInstance(jobConf, instance.getInstanceName()); GraphXInputFormat.setConnectorInfo(jobConf, username, password); GraphXInputFormat.setInputTableName(jobConf, table); GraphXInputFormat.setInputTableName(jobConf, table); GraphXInputFormat.setScanIsolation(jobConf, false); GraphXInputFormat.setLocalIterators(jobConf, false); GraphXInputFormat.setOfflineTableScan(jobConf, false); final GraphXInputFormat inputFormat = new GraphXInputFormat(); final JobContext context = new JobContextImpl(jobConf.getConfiguration(), jobConf.getJobID()); final List<InputSplit> splits = inputFormat.getSplits(context); Assert.assertEquals(1, splits.size()); final TaskAttemptContext taskAttemptContext = new TaskAttemptContextImpl(context.getConfiguration(), new TaskAttemptID(new TaskID(), 1)); final RecordReader<Object, RyaTypeWritable> reader = inputFormat.createRecordReader(splits.get(0), taskAttemptContext); final RyaStatementRecordReader ryaStatementRecordReader = (RyaStatementRecordReader)reader; ryaStatementRecordReader.initialize(splits.get(0), taskAttemptContext); final List<RyaType> results = new ArrayList<RyaType>(); System.out.println("before while"); while(ryaStatementRecordReader.nextKeyValue()) { System.out.println("in while"); final RyaTypeWritable writable = ryaStatementRecordReader.getCurrentValue(); final RyaType value = writable.getRyaType(); final Object text = ryaStatementRecordReader.getCurrentKey(); final RyaType type = new RyaType(); final String validatedLanguage = LiteralLanguageUtils.validateLanguage(value.getLanguage(), value.getDataType()); type.setData(value.getData()); type.setDataType(value.getDataType()); type.setLanguage(validatedLanguage); results.add(type); System.out.println(value.getData()); System.out.println(value.getDataType()); System.out.println(results); System.out.println(type); System.out.println(text); System.out.println(value); } System.out.println("after while"); System.out.println(results.size()); System.out.println(results); }
@Override public RecordReader<Object, RyaTypeWritable> createRecordReader( InputSplit split, TaskAttemptContext context) { return new RyaStatementRecordReader(); }
GraphXInputFormat extends InputFormatBase<Object, RyaTypeWritable> { @Override public RecordReader<Object, RyaTypeWritable> createRecordReader( InputSplit split, TaskAttemptContext context) { return new RyaStatementRecordReader(); } }
GraphXInputFormat extends InputFormatBase<Object, RyaTypeWritable> { @Override public RecordReader<Object, RyaTypeWritable> createRecordReader( InputSplit split, TaskAttemptContext context) { return new RyaStatementRecordReader(); }  }
GraphXInputFormat extends InputFormatBase<Object, RyaTypeWritable> { @Override public RecordReader<Object, RyaTypeWritable> createRecordReader( InputSplit split, TaskAttemptContext context) { return new RyaStatementRecordReader(); }  @Override RecordReader<Object, RyaTypeWritable> createRecordReader(
			InputSplit split, TaskAttemptContext context); }
GraphXInputFormat extends InputFormatBase<Object, RyaTypeWritable> { @Override public RecordReader<Object, RyaTypeWritable> createRecordReader( InputSplit split, TaskAttemptContext context) { return new RyaStatementRecordReader(); }  @Override RecordReader<Object, RyaTypeWritable> createRecordReader(
			InputSplit split, TaskAttemptContext context);  }