src/main/java/org/novus/service/IngestorServiceImpl.java
/**
   * This method parses CSV and creates a json out of it and feeds to ES CSV Should follow same order for
   * columns - As this is for Demo, there is no validation handled.
   * 
   * @param client
   * @param uploadingFiles
   * @throws IllegalStateException
   * @throws IOException
   */
This method parses CSV and creates a json out of it and feeds to ES CSV Should follow same order for
columns - As this is for Demo, there is no validation handled.

@param client
@param uploadingFiles
@throws IllegalStateException
@throws IOException
This method parses CSV and creates a json out of it and feeds to ES CSV Should follow same order for columns - As this is for Demo there is no validation handled . @param client @param uploadingFiles @throws IllegalStateException @throws IOException
public void csvLoader(String index, String type, TransportClient client, MultipartFile[] uploadingFiles)
      throws IllegalStateException, IOException {
    BulkRequestBuilder bulkRequest = client.prepareBulk();
    for (MultipartFile uploadedFile : uploadingFiles) {
      File file = new File(uploadDir + File.separator + uploadedFile.getOriginalFilename());
      if (!file.getName().endsWith("csv")) {
        throw new IllegalArgumentException("Invalid csv");
      }
      uploadedFile.transferTo(file);
      String line;
      int lineNumber = 1;
      String delimiter = ",";
      boolean isHeader = true;
      Map<Integer, Object> headers = new HashMap<>();
      try (BufferedReader br = new BufferedReader(new FileReader(file))) {
        while ((line = br.readLine()) != null) {
          String[] tokens = line.split(delimiter);
          Map<String, String> values = new HashMap<>();
          if (isHeader) {
            for (int i = 0; i < tokens.length; i++) {
              headers.put(i, tokens[i]);
            }
            isHeader = false;
          } else {
            for (int i = 0; i < tokens.length; i++) {
              try {
                values.put(headers.get(i).toString(), tokens[i]);
              } catch (Exception ex) {
                System.out.println(" Please check line number " + lineNumber
                    + " for comma or any ther format error with data " + tokens);
              }
            }
          }
          if (!CollectionUtils.isEmpty(values)) {
            String json = new ObjectMapper().writeValueAsString(values);
            bulkRequest.add(client.prepareIndex(index, type, tokens[0].toString()).setSource(json, XContentType.JSON));
          }
          lineNumber++;
          if ((lineNumber % 5000) == 0) {
            System.out.println("executing bulk updates lineNumber " + lineNumber);
            BulkResponse bulkResponse = bulkRequest.execute().actionGet();
            if (bulkResponse.hasFailures()) {
              System.out.println(bulkResponse.buildFailureMessage());
              log.error("============================== failures");
            }
            bulkRequest = client.prepareBulk();
          }
        }
        bulkRequest.setRefreshPolicy("true");
        final BulkResponse bulkResponse = bulkRequest.execute().actionGet();
        if (bulkResponse.hasFailures()) {
          log.error("============================== failures");
        }
      } catch (IOException ex) {
        ex.printStackTrace();
      }
    }
  }
public void csvLoader ( String index , String type , TransportClient client , MultipartFile [ ] uploadingFiles ) throws IllegalStateException , IOException { BulkRequestBuilder bulkRequest = client . prepareBulk ( ) ; for ( MultipartFile uploadedFile : uploadingFiles ) { File file = new File ( uploadDir + File . separator + uploadedFile . getOriginalFilename ( ) ) ; if ( ! file . getName ( ) . endsWith ( "csv" ) ) { throw new IllegalArgumentException ( "Invalid csv" ) ; } uploadedFile . transferTo ( file ) ; String line ; int lineNumber = 1 ; String delimiter = "," ; boolean isHeader = true ; Map < Integer , Object > headers = new HashMap < > ( ) ; try ( BufferedReader br = new BufferedReader ( new FileReader ( file ) ) ) { while ( ( line = br . readLine ( ) ) != null ) { String [ ] tokens = line . split ( delimiter ) ; Map < String , String > values = new HashMap < > ( ) ; if ( isHeader ) { for ( int i = 0 ; i < tokens . length ; i ++ ) { headers . put ( i , tokens [ i ] ) ; } isHeader = false ; } else { for ( int i = 0 ; i < tokens . length ; i ++ ) { try { values . put ( headers . get ( i ) . toString ( ) , tokens [ i ] ) ; } catch ( Exception ex ) { System . out . println ( " Please check line number " + lineNumber + " for comma or any ther format error with data " + tokens ) ; } } } if ( ! CollectionUtils . isEmpty ( values ) ) { String json = new ObjectMapper ( ) . writeValueAsString ( values ) ; bulkRequest . add ( client . prepareIndex ( index , type , tokens [ 0 ] . toString ( ) ) . setSource ( json , XContentType . JSON ) ) ; } lineNumber ++ ; if ( ( lineNumber % 5000 ) == 0 ) { System . out . println ( "executing bulk updates lineNumber " + lineNumber ) ; BulkResponse bulkResponse = bulkRequest . execute ( ) . actionGet ( ) ; if ( bulkResponse . hasFailures ( ) ) { System . out . println ( bulkResponse . buildFailureMessage ( ) ) ; log . error ( "============================== failures" ) ; } bulkRequest = client . prepareBulk ( ) ; } } bulkRequest . setRefreshPolicy ( "true" ) ; final BulkResponse bulkResponse = bulkRequest . execute ( ) . actionGet ( ) ; if ( bulkResponse . hasFailures ( ) ) { log . error ( "============================== failures" ) ; } } catch ( IOException ex ) { ex . printStackTrace ( ) ; } } }
This method parses CSV and creates a json out of it and feeds to ES CSV Should follow same order for
 columns - As this is for Demo, there is no validation handled.
This method parses CSV and creates a json out of it and feeds to ES CSV Should follow same order for columns - As this is for Demo there is no validation handled .
// ex.printStackTrace(); // System.out.println(line);