ElasticsearchScanner implements Scanner { public ElasticsearchScanner (Configuration conf, Schema schema, TableMeta tableMeta, Fragment fragment) throws IOException { this.conf = (TajoConf)conf; this.schema = schema; this.tableMeta = tableMeta; this.fragment = (ElasticsearchFragment)fragment; this.tableStats = new TableStats(); this.optionInfo = new ElasticsearchWithOptionInfo(tableMeta); this.client = null; } ElasticsearchScanner(Configuration conf, Schema schema, TableMeta tableMeta, Fragment fragment); @Override void init(); void documentScanner(); void setClient(Client client); @Override Tuple next(); @Override void reset(); @Override void close(); @Override boolean isProjectable(); @Override void setTarget(Column[] targets); @Override boolean isSelectable(); @Override void setSearchCondition(Object expr); @Override boolean isSplittable(); @Override float getProgress(); @Override TableStats getInputStats(); @Override Schema getSchema();  }
@Test public void testElasticsearchScanner() throws Exception { try { client.admin().indices().prepareDelete("test-ndex").execute().actionGet(); } catch (Exception e) { } finally { } Settings indexSettings = ImmutableSettings.settingsBuilder() .put("number_of_shards","1") .put("number_of_replicas", "0") .build(); XContentBuilder builder = XContentFactory.jsonBuilder() .startObject() .startObject("test-type") .startObject("_all") .field("enabled", "false") .endObject() .startObject("_id") .field("path", "field1") .endObject() .startObject("properties") .startObject("field1") .field("type", "long").field("store", "no").field("index", "not_analyzed") .endObject() .startObject("field2") .field("type", "string").field("store", "no").field("index", "not_analyzed") .endObject() .startObject("field3") .field("type", "string").field("store", "no").field("index", "analyzed") .endObject() .endObject() .endObject() .endObject(); CreateIndexResponse res = client.admin().indices().prepareCreate("test-index") .setSettings(indexSettings) .addMapping("test-type", builder) .execute() .actionGet(); assertEquals(res.isAcknowledged(), true); IndexRequestBuilder indexRequestBuilder = client.prepareIndex().setIndex("test-index").setType("test-type"); IndexResponse indexResponse; for ( int i=0; i<10; i++ ) { builder = XContentFactory.jsonBuilder() .startObject() .field("field1", i).field("field2", "henry" + i).field("field3", i + ". hello world!! elasticsearch on apache tajo!!") .endObject(); indexResponse = indexRequestBuilder.setSource(builder) .setId(String.valueOf(i)) .setOperationThreaded(false) .setConsistencyLevel(WriteConsistencyLevel.QUORUM) .setReplicationType(ReplicationType.ASYNC) .execute() .actionGet(); assertEquals(indexResponse.isCreated(), true); } client.admin().indices().prepareRefresh("test-index").execute().actionGet(); conf = new TajoConf(); schema = new Schema(); schema.addColumn("_type", TajoDataTypes.Type.TEXT); schema.addColumn("_score", TajoDataTypes.Type.FLOAT8); schema.addColumn("_id", TajoDataTypes.Type.TEXT); schema.addColumn("field1", TajoDataTypes.Type.INT8); schema.addColumn("field2", TajoDataTypes.Type.TEXT); schema.addColumn("field3", TajoDataTypes.Type.TEXT); Schema targetSchema = new Schema(); targetSchema.addColumn(schema.getColumn(0)); targetSchema.addColumn(schema.getColumn(1)); targetSchema.addColumn(schema.getColumn(2)); targetSchema.addColumn(schema.getColumn(3)); targetSchema.addColumn(schema.getColumn(4)); targetSchema.addColumn(schema.getColumn(5)); options = new KeyValueSet(); options.set("es.cluster", "testTajoCluster"); options.set("es.nodes", "localhost:9300"); options.set("es.index", "test-index"); options.set("es.type", "test-type"); tableMeta = CatalogUtil.newTableMeta(storeType, options); fragment = new ElasticsearchFragment(); fragment.setIndexName("test-index"); fragment.setIndexType("test-type"); fragment.setOffset(0); fragment.setFetchSize(10); fragment.setShardId(0); ElasticsearchScanner scanner = new ElasticsearchScanner(conf, targetSchema, tableMeta, fragment); scanner.setClient(client); scanner.init(); int totalCounts = 0; Tuple tuple; while ((tuple = scanner.next()) != null) { assertEquals("henry"+totalCounts, tuple.getText(4)); totalCounts++; } scanner.close(); assertEquals(totalCounts, 10); }